Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Bump pydantic from 1.10.7 to 2.0.1 #78

Closed
wants to merge 1 commit into from

Bump pydantic from 1.10.7 to 2.0.1

ee3aaf0
Select commit
Loading
Failed to load commit list.
Sign in for the full log view
Closed

Bump pydantic from 1.10.7 to 2.0.1 #78

Bump pydantic from 1.10.7 to 2.0.1
ee3aaf0
Select commit
Loading
Failed to load commit list.
This check has been archived and is scheduled for deletion. Learn more about checks retention
GitHub Actions / JUnit Test Report failed Jul 4, 2023 in 0s

25 tests run, 15 passed, 0 skipped, 10 failed.

Annotations

Check failure on line 88 in tests/test_to_avro.py

See this annotation in the file changed.

@github-actions 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:88: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
src/pydantic_avro/base.py:24: in avro_schema
    return cls._avro_schema(schema, namespace)
src/pydantic_avro/base.py:164: in _avro_schema
    fields = get_fields(schema)
src/pydantic_avro/base.py:151: 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"]
            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:139: NotImplementedError

Check failure on line 125 in tests/test_to_avro.py

See this annotation in the file changed.

@github-actions 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.0.2/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.0.2/v/string_type
Raw 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.0.2/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.0.2/v/string_type

tests/test_to_avro.py:125: ValidationError

Check failure on line 163 in tests/test_to_avro.py

See this annotation in the file changed.

@github-actions 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:163: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
src/pydantic_avro/base.py:24: in avro_schema
    return cls._avro_schema(schema, namespace)
src/pydantic_avro/base.py:164: in _avro_schema
    fields = get_fields(schema)
src/pydantic_avro/base.py:151: 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 181 in tests/test_to_avro.py

See this annotation in the file changed.

@github-actions 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:181: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
src/pydantic_avro/base.py:24: in avro_schema
    return cls._avro_schema(schema, namespace)
src/pydantic_avro/base.py:164: in _avro_schema
    fields = get_fields(schema)
src/pydantic_avro/base.py:151: 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 202 in tests/test_to_avro.py

See this annotation in the file changed.

@github-actions 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:202: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
src/pydantic_avro/base.py:24: in avro_schema
    return cls._avro_schema(schema, namespace)
src/pydantic_avro/base.py:164: in _avro_schema
    fields = get_fields(schema)
src/pydantic_avro/base.py:151: 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 246 in tests/test_to_avro.py

See this annotation in the file changed.

@github-actions 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.0.2/v/missing
Raw 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.0.2/v/missing

tests/test_to_avro.py:246: ValidationError

Check failure on line 275 in tests/test_to_avro.py

See this annotation in the file changed.

@github-actions 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:275: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
src/pydantic_avro/base.py:24: in avro_schema
    return cls._avro_schema(schema, namespace)
src/pydantic_avro/base.py:164: in _avro_schema
    fields = get_fields(schema)
src/pydantic_avro/base.py:151: 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"]
            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:139: NotImplementedError

Check failure on line 294 in tests/test_to_avro.py

See this annotation in the file changed.

@github-actions 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:294: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
src/pydantic_avro/base.py:24: in avro_schema
    return cls._avro_schema(schema, namespace)
src/pydantic_avro/base.py:164: in _avro_schema
    fields = get_fields(schema)
src/pydantic_avro/base.py:151: 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"]
            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:139: NotImplementedError

Check failure on line 323 in tests/test_to_avro.py

See this annotation in the file changed.

@github-actions 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:323: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
src/pydantic_avro/base.py:24: in avro_schema
    return cls._avro_schema(schema, namespace)
src/pydantic_avro/base.py:164: in _avro_schema
    fields = get_fields(schema)
src/pydantic_avro/base.py:151: 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 381 in tests/test_to_avro.py

See this annotation in the file changed.

@github-actions 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:381: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
src/pydantic_avro/base.py:24: in avro_schema
    return cls._avro_schema(schema, namespace)
src/pydantic_avro/base.py:164: in _avro_schema
    fields = get_fields(schema)
src/pydantic_avro/base.py:151: 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"]
            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:139: NotImplementedError
View more details on GitHub Actions