Bump pydantic from 1.10.7 to 2.1.0 #246
GitHub Actions / JUnit Test Report
failed
Jul 25, 2023 in 0s
26 tests run, 15 passed, 0 skipped, 11 failed.
Annotations
Check failure on line 94 in tests/test_to_avro.py
github-actions / JUnit Test Report
test_to_avro.test_avro
NotImplementedError: Type 'null' not support yet, please report this at https://github.com/godatadriven/pydantic-avro/issues
Raw output
def test_avro():
> result = TestModel.avro_schema()
tests/test_to_avro.py:94:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
src/pydantic_avro/base.py:24: in avro_schema
return cls._avro_schema(schema, namespace)
src/pydantic_avro/base.py:172: in _avro_schema
fields = get_fields(schema)
src/pydantic_avro/base.py:159: in get_fields
avro_type_dict = get_type(value)
src/pydantic_avro/base.py:59: in get_type
avro_type_dict["type"].append(get_type(union_element)["type"])
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
value = {'type': 'null'}
def get_type(value: dict) -> dict:
"""Returns a type of a single field"""
t = value.get("type")
f = value.get("format")
r = value.get("$ref")
a = value.get("additionalProperties")
u = value.get("anyOf")
minimum = value.get("minimum")
maximum = value.get("maximum")
avro_type_dict: Dict[str, Any] = {}
if "default" in value:
avro_type_dict["default"] = value.get("default")
if "description" in value:
avro_type_dict["doc"] = value.get("description")
if "allOf" in value and len(value["allOf"]) == 1:
r = value["allOf"][0]["$ref"]
if u is not None:
avro_type_dict["type"] = []
for union_element in u:
avro_type_dict["type"].append(get_type(union_element)["type"])
elif r is not None:
class_name = r.replace("#/definitions/", "")
if class_name in classes_seen:
avro_type_dict["type"] = class_name
else:
d = get_definition(r, schema)
if "enum" in d:
avro_type_dict["type"] = {
"type": "enum",
"symbols": [str(v) for v in d["enum"]],
"name": d["title"],
}
else:
avro_type_dict["type"] = {
"type": "record",
"fields": get_fields(d),
# Name of the struct should be unique true the complete schema
# Because of this the path in the schema is tracked and used as name for a nested struct/array
"name": class_name,
}
classes_seen.add(class_name)
elif t == "array":
items = value.get("items")
tn = get_type(items)
# If items in array are a object:
if "$ref" in items:
tn = tn["type"]
# If items in array are a logicalType
if (
isinstance(tn, dict)
and isinstance(tn.get("type", {}), dict)
and tn.get("type", {}).get("logicalType") is not None
):
tn = tn["type"]
# If items in array are an array, the structure must be corrected
if (
isinstance(tn, dict)
and isinstance(tn.get("type", {}), dict)
and tn.get("type", {}).get("type") == "array"
):
items = tn["type"]["items"]
tn = {"type": "array", "items": items}
avro_type_dict["type"] = {"type": "array", "items": tn}
elif t == "string" and f == "date-time":
avro_type_dict["type"] = {
"type": "long",
"logicalType": "timestamp-micros",
}
elif t == "string" and f == "date":
avro_type_dict["type"] = {
"type": "int",
"logicalType": "date",
}
elif t == "string" and f == "time":
avro_type_dict["type"] = {
"type": "long",
"logicalType": "time-micros",
}
elif t == "string" and f == "uuid":
avro_type_dict["type"] = {
"type": "string",
"logicalType": "uuid",
}
elif t == "string" and f == "binary":
avro_type_dict["type"] = "bytes"
elif t == "string":
avro_type_dict["type"] = "string"
elif t == "number":
avro_type_dict["type"] = "double"
elif t == "integer":
# integer in python can be a long, only if minimum and maximum value is set a int can be used
if minimum is not None and minimum >= -(2**31) and maximum is not None and maximum <= (2**31 - 1):
avro_type_dict["type"] = "int"
else:
avro_type_dict["type"] = "long"
elif t == "boolean":
avro_type_dict["type"] = "boolean"
elif t == "object":
if a is None:
value_type = "string"
else:
value_type = get_type(a)
if isinstance(value_type, dict) and len(value_type) == 1:
value_type = value_type.get("type")
avro_type_dict["type"] = {"type": "map", "values": value_type}
else:
> raise NotImplementedError(
f"Type '{t}' not support yet, "
f"please report this at https://github.com/godatadriven/pydantic-avro/issues"
)
E NotImplementedError: Type 'null' not support yet, please report this at https://github.com/godatadriven/pydantic-avro/issues
src/pydantic_avro/base.py:147: NotImplementedError
Check failure on line 131 in tests/test_to_avro.py
github-actions / JUnit Test Report
test_to_avro.test_avro_write
pydantic_core._pydantic_core.ValidationError: 2 validation errors for TestModel
c5
Datetimes provided to dates should have zero time - e.g. be exact dates [type=date_from_datetime_inexact, input_value=5, input_type=int]
For further information visit https://errors.pydantic.dev/2.1/v/date_from_datetime_inexact
c7
Input should be a valid string [type=string_type, input_value=7, input_type=int]
For further information visit https://errors.pydantic.dev/2.1/v/string_typeRaw output
def test_avro_write():
> record1 = TestModel(
c1="1",
c2=2,
c3=3,
c4=4,
c5=5,
c6=6,
c7=7,
c8=True,
c9=uuid.uuid4(),
c10=uuid.uuid4(),
c11={"key": "value"},
c12={},
c13=Status.passed,
c14=bytes(),
)
E pydantic_core._pydantic_core.ValidationError: 2 validation errors for TestModel
E c5
E Datetimes provided to dates should have zero time - e.g. be exact dates [type=date_from_datetime_inexact, input_value=5, input_type=int]
E For further information visit https://errors.pydantic.dev/2.1/v/date_from_datetime_inexact
E c7
E Input should be a valid string [type=string_type, input_value=7, input_type=int]
E For further information visit https://errors.pydantic.dev/2.1/v/string_type
tests/test_to_avro.py:131: ValidationError
Check failure on line 169 in tests/test_to_avro.py
github-actions / JUnit Test Report
test_to_avro.test_reused_object
RuntimeError: Definition #/$defs/Nested2Model does not exist
Raw output
def test_reused_object():
> result = ReusedObject.avro_schema()
tests/test_to_avro.py:169:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
src/pydantic_avro/base.py:24: in avro_schema
return cls._avro_schema(schema, namespace)
src/pydantic_avro/base.py:172: in _avro_schema
fields = get_fields(schema)
src/pydantic_avro/base.py:159: in get_fields
avro_type_dict = get_type(value)
src/pydantic_avro/base.py:65: in get_type
d = get_definition(r, schema)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
ref = '#/$defs/Nested2Model'
schema = {'$defs': {'Nested2Model': {'properties': {'c111': {'title': 'C111', 'type': 'string'}}, 'required': ['c111'], 'title'.../$defs/Nested2Model'}, 'c2': {'$ref': '#/$defs/Nested2Model'}}, 'required': ['c1', 'c2'], 'title': 'ReusedObject', ...}
def get_definition(ref: str, schema: dict):
"""Reading definition of base schema for nested structs"""
id = ref.replace("#/definitions/", "")
d = schema.get("definitions", {}).get(id)
if d is None:
> raise RuntimeError(f"Definition {id} does not exist")
E RuntimeError: Definition #/$defs/Nested2Model does not exist
src/pydantic_avro/base.py:37: RuntimeError
Check failure on line 187 in tests/test_to_avro.py
github-actions / JUnit Test Report
test_to_avro.test_reused_object_array
RuntimeError: Definition #/$defs/Nested2Model does not exist
Raw output
def test_reused_object_array():
> result = ReusedObjectArray.avro_schema()
tests/test_to_avro.py:187:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
src/pydantic_avro/base.py:24: in avro_schema
return cls._avro_schema(schema, namespace)
src/pydantic_avro/base.py:172: in _avro_schema
fields = get_fields(schema)
src/pydantic_avro/base.py:159: in get_fields
avro_type_dict = get_type(value)
src/pydantic_avro/base.py:84: in get_type
tn = get_type(items)
src/pydantic_avro/base.py:65: in get_type
d = get_definition(r, schema)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
ref = '#/$defs/Nested2Model'
schema = {'$defs': {'Nested2Model': {'properties': {'c111': {'title': 'C111', 'type': 'string'}}, 'required': ['c111'], 'title'...'type': 'array'}, 'c2': {'$ref': '#/$defs/Nested2Model'}}, 'required': ['c1', 'c2'], 'title': 'ReusedObjectArray', ...}
def get_definition(ref: str, schema: dict):
"""Reading definition of base schema for nested structs"""
id = ref.replace("#/definitions/", "")
d = schema.get("definitions", {}).get(id)
if d is None:
> raise RuntimeError(f"Definition {id} does not exist")
E RuntimeError: Definition #/$defs/Nested2Model does not exist
src/pydantic_avro/base.py:37: RuntimeError
Check failure on line 208 in tests/test_to_avro.py
github-actions / JUnit Test Report
test_to_avro.test_complex_avro
RuntimeError: Definition #/$defs/NestedModel does not exist
Raw output
def test_complex_avro():
> result = ComplexTestModel.avro_schema()
tests/test_to_avro.py:208:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
src/pydantic_avro/base.py:24: in avro_schema
return cls._avro_schema(schema, namespace)
src/pydantic_avro/base.py:172: in _avro_schema
fields = get_fields(schema)
src/pydantic_avro/base.py:159: in get_fields
avro_type_dict = get_type(value)
src/pydantic_avro/base.py:65: in get_type
d = get_definition(r, schema)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
ref = '#/$defs/NestedModel'
schema = {'$defs': {'Nested2Model': {'properties': {'c111': {'title': 'C111', 'type': 'string'}}, 'required': ['c111'], 'title'...tle': 'C4', 'type': 'array'}, ...}, 'required': ['c1', 'c2', 'c3', 'c4', 'c5', 'c6'], 'title': 'ComplexTestModel', ...}
def get_definition(ref: str, schema: dict):
"""Reading definition of base schema for nested structs"""
id = ref.replace("#/definitions/", "")
d = schema.get("definitions", {}).get(id)
if d is None:
> raise RuntimeError(f"Definition {id} does not exist")
E RuntimeError: Definition #/$defs/NestedModel does not exist
src/pydantic_avro/base.py:37: RuntimeError
Check failure on line 271 in tests/test_to_avro.py
github-actions / JUnit Test Report
test_to_avro.test_avro_parse_list_of_lists
AssertionError: assert [{'c1': 1, 'c... 5], [6, 7]]}] == [ListofLists(... 5], [6, 7]])]
At index 0 diff: {'c1': 1, 'c2': [2, 3], 'c3': [[4, 5], [6, 7]]} != ListofLists(c1=1, c2=[2, 3], c3=[[4, 5], [6, 7]])
Full diff:
- [ListofLists(c1=1, c2=[2, 3], c3=[[4, 5], [6, 7]])]
+ [{'c1': 1, 'c2': [2, 3], 'c3': [[4, 5], [6, 7]]}]Raw output
def test_avro_parse_list_of_lists():
record = ListofLists(c1=1, c2=[2, 3], c3=[[4, 5], [6, 7]])
schema = ListofLists.avro_schema()
parsed_schema = parse_schema(schema)
records = [
record.dict(),
]
with tempfile.TemporaryDirectory() as dir:
# Writing
with open(os.path.join(dir, "test.avro"), "wb") as out:
writer(out, parsed_schema, records)
result_records = []
# Reading
with open(os.path.join(dir, "test.avro"), "rb") as fo:
for record in reader(fo):
result_records.append(ListofLists.parse_obj(record))
> assert records == result_records
E AssertionError: assert [{'c1': 1, 'c... 5], [6, 7]]}] == [ListofLists(... 5], [6, 7]])]
E At index 0 diff: {'c1': 1, 'c2': [2, 3], 'c3': [[4, 5], [6, 7]]} != ListofLists(c1=1, c2=[2, 3], c3=[[4, 5], [6, 7]])
E Full diff:
E - [ListofLists(c1=1, c2=[2, 3], c3=[[4, 5], [6, 7]])]
E + [{'c1': 1, 'c2': [2, 3], 'c3': [[4, 5], [6, 7]]}]
tests/test_to_avro.py:271: AssertionError
Check failure on line 275 in tests/test_to_avro.py
github-actions / JUnit Test Report
test_to_avro.test_avro_write_complex
pydantic_core._pydantic_core.ValidationError: 1 validation error for ComplexTestModel
c6
Field required [type=missing, input_value={'c1': ['1', '2'], 'c2': ...ed2Model(c111='test'))}}, input_type=dict]
For further information visit https://errors.pydantic.dev/2.1/v/missingRaw output
def test_avro_write_complex():
> record1 = ComplexTestModel(
c1=["1", "2"],
c2=NestedModel(c11=Nested2Model(c111="test")),
c3=[NestedModel(c11=Nested2Model(c111="test"))],
c4=[1, 2, 3, 4],
c5={"key": NestedModel(c11=Nested2Model(c111="test"))},
)
E pydantic_core._pydantic_core.ValidationError: 1 validation error for ComplexTestModel
E c6
E Field required [type=missing, input_value={'c1': ['1', '2'], 'c2': ...ed2Model(c111='test'))}}, input_type=dict]
E For further information visit https://errors.pydantic.dev/2.1/v/missing
tests/test_to_avro.py:275: ValidationError
Check failure on line 304 in tests/test_to_avro.py
github-actions / JUnit Test Report
test_to_avro.test_defaults
NotImplementedError: Type 'null' not support yet, please report this at https://github.com/godatadriven/pydantic-avro/issues
Raw output
def test_defaults():
> result = DefaultValues.avro_schema()
tests/test_to_avro.py:304:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
src/pydantic_avro/base.py:24: in avro_schema
return cls._avro_schema(schema, namespace)
src/pydantic_avro/base.py:172: in _avro_schema
fields = get_fields(schema)
src/pydantic_avro/base.py:159: in get_fields
avro_type_dict = get_type(value)
src/pydantic_avro/base.py:59: in get_type
avro_type_dict["type"].append(get_type(union_element)["type"])
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
value = {'type': 'null'}
def get_type(value: dict) -> dict:
"""Returns a type of a single field"""
t = value.get("type")
f = value.get("format")
r = value.get("$ref")
a = value.get("additionalProperties")
u = value.get("anyOf")
minimum = value.get("minimum")
maximum = value.get("maximum")
avro_type_dict: Dict[str, Any] = {}
if "default" in value:
avro_type_dict["default"] = value.get("default")
if "description" in value:
avro_type_dict["doc"] = value.get("description")
if "allOf" in value and len(value["allOf"]) == 1:
r = value["allOf"][0]["$ref"]
if u is not None:
avro_type_dict["type"] = []
for union_element in u:
avro_type_dict["type"].append(get_type(union_element)["type"])
elif r is not None:
class_name = r.replace("#/definitions/", "")
if class_name in classes_seen:
avro_type_dict["type"] = class_name
else:
d = get_definition(r, schema)
if "enum" in d:
avro_type_dict["type"] = {
"type": "enum",
"symbols": [str(v) for v in d["enum"]],
"name": d["title"],
}
else:
avro_type_dict["type"] = {
"type": "record",
"fields": get_fields(d),
# Name of the struct should be unique true the complete schema
# Because of this the path in the schema is tracked and used as name for a nested struct/array
"name": class_name,
}
classes_seen.add(class_name)
elif t == "array":
items = value.get("items")
tn = get_type(items)
# If items in array are a object:
if "$ref" in items:
tn = tn["type"]
# If items in array are a logicalType
if (
isinstance(tn, dict)
and isinstance(tn.get("type", {}), dict)
and tn.get("type", {}).get("logicalType") is not None
):
tn = tn["type"]
# If items in array are an array, the structure must be corrected
if (
isinstance(tn, dict)
and isinstance(tn.get("type", {}), dict)
and tn.get("type", {}).get("type") == "array"
):
items = tn["type"]["items"]
tn = {"type": "array", "items": items}
avro_type_dict["type"] = {"type": "array", "items": tn}
elif t == "string" and f == "date-time":
avro_type_dict["type"] = {
"type": "long",
"logicalType": "timestamp-micros",
}
elif t == "string" and f == "date":
avro_type_dict["type"] = {
"type": "int",
"logicalType": "date",
}
elif t == "string" and f == "time":
avro_type_dict["type"] = {
"type": "long",
"logicalType": "time-micros",
}
elif t == "string" and f == "uuid":
avro_type_dict["type"] = {
"type": "string",
"logicalType": "uuid",
}
elif t == "string" and f == "binary":
avro_type_dict["type"] = "bytes"
elif t == "string":
avro_type_dict["type"] = "string"
elif t == "number":
avro_type_dict["type"] = "double"
elif t == "integer":
# integer in python can be a long, only if minimum and maximum value is set a int can be used
if minimum is not None and minimum >= -(2**31) and maximum is not None and maximum <= (2**31 - 1):
avro_type_dict["type"] = "int"
else:
avro_type_dict["type"] = "long"
elif t == "boolean":
avro_type_dict["type"] = "boolean"
elif t == "object":
if a is None:
value_type = "string"
else:
value_type = get_type(a)
if isinstance(value_type, dict) and len(value_type) == 1:
value_type = value_type.get("type")
avro_type_dict["type"] = {"type": "map", "values": value_type}
else:
> raise NotImplementedError(
f"Type '{t}' not support yet, "
f"please report this at https://github.com/godatadriven/pydantic-avro/issues"
)
E NotImplementedError: Type 'null' not support yet, please report this at https://github.com/godatadriven/pydantic-avro/issues
src/pydantic_avro/base.py:147: NotImplementedError
Check failure on line 323 in tests/test_to_avro.py
github-actions / JUnit Test Report
test_to_avro.test_custom_namespace
NotImplementedError: Type 'null' not support yet, please report this at https://github.com/godatadriven/pydantic-avro/issues
Raw output
def test_custom_namespace():
# if given namespace should change namespace in avro schema
> result = DefaultValues.avro_schema(namespace="test.test")
tests/test_to_avro.py:323:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
src/pydantic_avro/base.py:24: in avro_schema
return cls._avro_schema(schema, namespace)
src/pydantic_avro/base.py:172: in _avro_schema
fields = get_fields(schema)
src/pydantic_avro/base.py:159: in get_fields
avro_type_dict = get_type(value)
src/pydantic_avro/base.py:59: in get_type
avro_type_dict["type"].append(get_type(union_element)["type"])
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
value = {'type': 'null'}
def get_type(value: dict) -> dict:
"""Returns a type of a single field"""
t = value.get("type")
f = value.get("format")
r = value.get("$ref")
a = value.get("additionalProperties")
u = value.get("anyOf")
minimum = value.get("minimum")
maximum = value.get("maximum")
avro_type_dict: Dict[str, Any] = {}
if "default" in value:
avro_type_dict["default"] = value.get("default")
if "description" in value:
avro_type_dict["doc"] = value.get("description")
if "allOf" in value and len(value["allOf"]) == 1:
r = value["allOf"][0]["$ref"]
if u is not None:
avro_type_dict["type"] = []
for union_element in u:
avro_type_dict["type"].append(get_type(union_element)["type"])
elif r is not None:
class_name = r.replace("#/definitions/", "")
if class_name in classes_seen:
avro_type_dict["type"] = class_name
else:
d = get_definition(r, schema)
if "enum" in d:
avro_type_dict["type"] = {
"type": "enum",
"symbols": [str(v) for v in d["enum"]],
"name": d["title"],
}
else:
avro_type_dict["type"] = {
"type": "record",
"fields": get_fields(d),
# Name of the struct should be unique true the complete schema
# Because of this the path in the schema is tracked and used as name for a nested struct/array
"name": class_name,
}
classes_seen.add(class_name)
elif t == "array":
items = value.get("items")
tn = get_type(items)
# If items in array are a object:
if "$ref" in items:
tn = tn["type"]
# If items in array are a logicalType
if (
isinstance(tn, dict)
and isinstance(tn.get("type", {}), dict)
and tn.get("type", {}).get("logicalType") is not None
):
tn = tn["type"]
# If items in array are an array, the structure must be corrected
if (
isinstance(tn, dict)
and isinstance(tn.get("type", {}), dict)
and tn.get("type", {}).get("type") == "array"
):
items = tn["type"]["items"]
tn = {"type": "array", "items": items}
avro_type_dict["type"] = {"type": "array", "items": tn}
elif t == "string" and f == "date-time":
avro_type_dict["type"] = {
"type": "long",
"logicalType": "timestamp-micros",
}
elif t == "string" and f == "date":
avro_type_dict["type"] = {
"type": "int",
"logicalType": "date",
}
elif t == "string" and f == "time":
avro_type_dict["type"] = {
"type": "long",
"logicalType": "time-micros",
}
elif t == "string" and f == "uuid":
avro_type_dict["type"] = {
"type": "string",
"logicalType": "uuid",
}
elif t == "string" and f == "binary":
avro_type_dict["type"] = "bytes"
elif t == "string":
avro_type_dict["type"] = "string"
elif t == "number":
avro_type_dict["type"] = "double"
elif t == "integer":
# integer in python can be a long, only if minimum and maximum value is set a int can be used
if minimum is not None and minimum >= -(2**31) and maximum is not None and maximum <= (2**31 - 1):
avro_type_dict["type"] = "int"
else:
avro_type_dict["type"] = "long"
elif t == "boolean":
avro_type_dict["type"] = "boolean"
elif t == "object":
if a is None:
value_type = "string"
else:
value_type = get_type(a)
if isinstance(value_type, dict) and len(value_type) == 1:
value_type = value_type.get("type")
avro_type_dict["type"] = {"type": "map", "values": value_type}
else:
> raise NotImplementedError(
f"Type '{t}' not support yet, "
f"please report this at https://github.com/godatadriven/pydantic-avro/issues"
)
E NotImplementedError: Type 'null' not support yet, please report this at https://github.com/godatadriven/pydantic-avro/issues
src/pydantic_avro/base.py:147: NotImplementedError
Check failure on line 352 in tests/test_to_avro.py
github-actions / JUnit Test Report
test_to_avro.test_union_avro
RuntimeError: Definition #/$defs/NestedModel does not exist
Raw output
def test_union_avro():
> result = ModelWithUnion.avro_schema()
tests/test_to_avro.py:352:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
src/pydantic_avro/base.py:24: in avro_schema
return cls._avro_schema(schema, namespace)
src/pydantic_avro/base.py:172: in _avro_schema
fields = get_fields(schema)
src/pydantic_avro/base.py:159: in get_fields
avro_type_dict = get_type(value)
src/pydantic_avro/base.py:59: in get_type
avro_type_dict["type"].append(get_type(union_element)["type"])
src/pydantic_avro/base.py:65: in get_type
d = get_definition(r, schema)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
ref = '#/$defs/NestedModel'
schema = {'$defs': {'Nested2Model': {'properties': {'c111': {'title': 'C111', 'type': 'string'}}, 'required': ['c111'], 'title'...r'}, {'$ref': '#/$defs/NestedModel'}], 'title': 'C3'}}, 'required': ['c1', 'c2', 'c3'], 'title': 'ModelWithUnion', ...}
def get_definition(ref: str, schema: dict):
"""Reading definition of base schema for nested structs"""
id = ref.replace("#/definitions/", "")
d = schema.get("definitions", {}).get(id)
if d is None:
> raise RuntimeError(f"Definition {id} does not exist")
E RuntimeError: Definition #/$defs/NestedModel does not exist
src/pydantic_avro/base.py:37: RuntimeError
Check failure on line 410 in tests/test_to_avro.py
github-actions / JUnit Test Report
test_to_avro.test_optional_array
NotImplementedError: Type 'null' not support yet, please report this at https://github.com/godatadriven/pydantic-avro/issues
Raw output
def test_optional_array():
> result = OptionalArray.avro_schema()
tests/test_to_avro.py:410:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
src/pydantic_avro/base.py:24: in avro_schema
return cls._avro_schema(schema, namespace)
src/pydantic_avro/base.py:172: in _avro_schema
fields = get_fields(schema)
src/pydantic_avro/base.py:159: in get_fields
avro_type_dict = get_type(value)
src/pydantic_avro/base.py:59: in get_type
avro_type_dict["type"].append(get_type(union_element)["type"])
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
value = {'type': 'null'}
def get_type(value: dict) -> dict:
"""Returns a type of a single field"""
t = value.get("type")
f = value.get("format")
r = value.get("$ref")
a = value.get("additionalProperties")
u = value.get("anyOf")
minimum = value.get("minimum")
maximum = value.get("maximum")
avro_type_dict: Dict[str, Any] = {}
if "default" in value:
avro_type_dict["default"] = value.get("default")
if "description" in value:
avro_type_dict["doc"] = value.get("description")
if "allOf" in value and len(value["allOf"]) == 1:
r = value["allOf"][0]["$ref"]
if u is not None:
avro_type_dict["type"] = []
for union_element in u:
avro_type_dict["type"].append(get_type(union_element)["type"])
elif r is not None:
class_name = r.replace("#/definitions/", "")
if class_name in classes_seen:
avro_type_dict["type"] = class_name
else:
d = get_definition(r, schema)
if "enum" in d:
avro_type_dict["type"] = {
"type": "enum",
"symbols": [str(v) for v in d["enum"]],
"name": d["title"],
}
else:
avro_type_dict["type"] = {
"type": "record",
"fields": get_fields(d),
# Name of the struct should be unique true the complete schema
# Because of this the path in the schema is tracked and used as name for a nested struct/array
"name": class_name,
}
classes_seen.add(class_name)
elif t == "array":
items = value.get("items")
tn = get_type(items)
# If items in array are a object:
if "$ref" in items:
tn = tn["type"]
# If items in array are a logicalType
if (
isinstance(tn, dict)
and isinstance(tn.get("type", {}), dict)
and tn.get("type", {}).get("logicalType") is not None
):
tn = tn["type"]
# If items in array are an array, the structure must be corrected
if (
isinstance(tn, dict)
and isinstance(tn.get("type", {}), dict)
and tn.get("type", {}).get("type") == "array"
):
items = tn["type"]["items"]
tn = {"type": "array", "items": items}
avro_type_dict["type"] = {"type": "array", "items": tn}
elif t == "string" and f == "date-time":
avro_type_dict["type"] = {
"type": "long",
"logicalType": "timestamp-micros",
}
elif t == "string" and f == "date":
avro_type_dict["type"] = {
"type": "int",
"logicalType": "date",
}
elif t == "string" and f == "time":
avro_type_dict["type"] = {
"type": "long",
"logicalType": "time-micros",
}
elif t == "string" and f == "uuid":
avro_type_dict["type"] = {
"type": "string",
"logicalType": "uuid",
}
elif t == "string" and f == "binary":
avro_type_dict["type"] = "bytes"
elif t == "string":
avro_type_dict["type"] = "string"
elif t == "number":
avro_type_dict["type"] = "double"
elif t == "integer":
# integer in python can be a long, only if minimum and maximum value is set a int can be used
if minimum is not None and minimum >= -(2**31) and maximum is not None and maximum <= (2**31 - 1):
avro_type_dict["type"] = "int"
else:
avro_type_dict["type"] = "long"
elif t == "boolean":
avro_type_dict["type"] = "boolean"
elif t == "object":
if a is None:
value_type = "string"
else:
value_type = get_type(a)
if isinstance(value_type, dict) and len(value_type) == 1:
value_type = value_type.get("type")
avro_type_dict["type"] = {"type": "map", "values": value_type}
else:
> raise NotImplementedError(
f"Type '{t}' not support yet, "
f"please report this at https://github.com/godatadriven/pydantic-avro/issues"
)
E NotImplementedError: Type 'null' not support yet, please report this at https://github.com/godatadriven/pydantic-avro/issues
src/pydantic_avro/base.py:147: NotImplementedError
Loading