### UnionSchema example with default_factory Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/migration_guide.md Example demonstrating the usage of Unions with default_factory, highlighting the change in handling callables. ```python import dataclasses from dataclasses_avroschema import AvroModel @dataclasses.dataclass class Bus(AvroModel): "A Bus" engine_name: str class Meta: namespace = "types" @dataclasses.dataclass class Car(AvroModel): "A Car" engine_name: str class Meta: namespace = "types" @dataclasses.dataclass class UnionSchema(AvroModel): "Some Unions" lake_trip: Bus | Car river_trip: Bus | Car | None = None mountain_trip: Bus | Car = dataclasses.field( default_factory=lambda: Bus(engine_name="honda")) # mountain_trip: Bus | Car = dataclasses.field(default_factory=lambda: {"engine_name": "honda"}) # OLD WAY UnionSchema.avro_schema() ``` -------------------------------- ### schema evolution example Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/examples.md A synchronous example using kafka-python to demonstrate schema evolution with FULL compatibility. ```python --8<-- "examples/kafka-examples/schema-evolution-example/schema_evolution_example/app.py" ``` -------------------------------- ### kafka-python example Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/examples.md A synchronous example using kafka-python to demonstrate avro-json serialization. ```python --8<-- "examples/kafka-examples/kafka-python-example/kafka_python_example/app.py" ``` -------------------------------- ### rabbitmq example Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/examples.md A minimal example using RabbitMQ with the pika driver. ```python --8<-- "examples/rabbitmq-examples/rabbitmq-pika/rabbitmq_pika/app.py" ``` -------------------------------- ### Using namespaces for repeated types Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/migration_guide.md Demonstrates how to use namespaces to avoid redefining types when they are referenced multiple times, using a Trip and Location example. ```python import dataclasses import datetime from dataclasses_avroschema import AvroModel @dataclasses.dataclass class Location(AvroModel): latitude: float longitude: float class Meta: namespace = "types.location_type" # REQUIRED!!!! @dataclasses.dataclass class Trip(AvroModel): start_time: datetime.datetime start_location: Location finish_time: datetime.datetime finish_location: Location Trip.avro_schema_to_python() ``` -------------------------------- ### redis streams example Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/examples.md A minimal example using redis streams with the walrus driver. ```python --8<-- "examples/redis-examples/redis-streams-example/redis_streams_example/app.py" ``` -------------------------------- ### User schema in new version (>= 0.14.0) Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/migration_guide.md Example of the User schema after version 0.14.0, requiring inheritance from AvroModel and using the standard avro_schema() method. ```python import dataclasses import typing from dataclasses_avroschema import AvroModel, types @dataclasses.dataclass class User(AvroModel): "An User" name: str age: int pets: typing.List[str] accounts: typing.Dict[str, int] favorite_colors: types.Enum = types.Enum(["BLUE", "YELLOW", "GREEN"]) country: str = "Argentina" address: str = None User.avro_schema() ``` -------------------------------- ### Example of User schema with nested map Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/migration_guide.md This snippet shows how to define a User schema with a nested map for addresses, demonstrating the structure of Avro schemas generated by the library. ```python User.avro_schema() ``` -------------------------------- ### User schema in versions < 0.14.0 Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/migration_guide.md Example of how a User schema was defined before version 0.14.0, using SchemaGenerator and not inheriting from AvroModel. ```python # Versions < 0.14.0 import dataclasses import typing from dataclasses_avroschema import SchemaGenerator, types class User: "An User" name: str age: int pets: typing.List[str] accounts: typing.Dict[str, int] favorite_colors: types.Enum = types.Enum(["BLUE", "YELLOW", "GREEN"]) country: str = "Argentina" address: str = None SchemaGenerator(User).avro_schema() ``` -------------------------------- ### aiokafka example Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/examples.md An asynchronous example using aiokafka to serialize an AvroModel instance, send it through Kafka, and consume the event. ```python --8<-- "examples/kafka-examples/aiokafka-example/aiokafka_example/app.py" ``` -------------------------------- ### Json and Dict example Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/records.md Example demonstrating how to get JSON and dictionary representations of an AvroModel instance. ```python import dataclasses from dataclasses_avroschema import AvroModel @dataclasses.dataclass class User(AvroModel): """My User Class""" name: str age: int has_pets: bool = False money: float = 100.3 user = User(name="Bond", age=50) user.to_json() # >>> '{"name": "Bond", "age": 50, "has_pets": false, "money": 100.3}' user.to_dict() # >>> {'name': 'Bond', 'age': 50, 'has_pets': False, 'money': 100.3} ``` -------------------------------- ### Serialization Examples Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/README.md Examples showing how to serialize dataclass instances into Avro binary and Avro JSON formats. ```python assert user.serialize() == b"\x08john(\x02\x08test\x14\x00" ``` ```python assert user.serialize( serialization_type="avro-json" ) == b'{"name": "john", "age": 20, "addresses": [{"street": "test", "street_number": 10}]}' ``` ```python assert user.to_json() == '{"name": "john", "age": 20, "addresses": [{"street": "test", "street_number": 10}]}' ``` ```python assert user.to_dict() == { "name": "john", "age": 20, "addresses": [ {"street": "test", "street_number": 10} ] } ``` -------------------------------- ### Date example Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/logical_types.md Example demonstrating the use of the 'date' logical type with Python's datetime.date. ```python import datetime import dataclasses import typing from dataclasses_avroschema import AvroModel a_datetime = datetime.datetime(2019, 10, 12, 17, 57, 42) @dataclasses.dataclass class DateLogicalType(AvroModel): """Date type""" birthday: datetime.date meeting_date: typing.Optional[datetime.date] = None release_datetime: datetime.date = a_datetime.date() DateLogicalType.avro_schema() { "type": "record", "name": "DateLogicalType", "fields": [ { "name": "birthday", "type": {"type": "int", "logicalType": "date"}}, { "name": "meeting_date", "type": ["null", {"type": "int", "logicalType": "date"}], "default": null }, { "name": "release_datetime", "type": {"type": "int", "logicalType": "date"}, "default": 18181 } ], "doc": "Date type" } ``` -------------------------------- ### Timedelta example Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/logical_types.md Example demonstrating how timedelta fields are serialized to a double number of seconds in Avro. ```python import datetime import dataclasses from dataclasses_avroschema import AvroModel delta = datetime.timedelta(weeks=1, days=2, hours=3, minutes=4, seconds=5, milliseconds=6, microseconds=7) @dataclasses.dataclass class TimedeltaLogicalType(AvroModel): """Timedelta logical type""" time_elapsed: datetime.timedelta = delta DatetimeLogicalType.avro_schema() '{ "type": "record", "name": "DatetimeLogicalType", "fields": [ { "name": "time_elapsed", "type": { "type": "double", "logicalType": "dataclasses-avroschema-timedelta" }, "default": 788645.006007 } ], "doc": "Timedelta logical type" }' ``` -------------------------------- ### Nested record naming convention change Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/migration_guide.md Example showing the change in naming convention for nested records, from `class.__name__.lower()_record` to `class.__name__`. ```python import typing from dataclasses_avroschema import AvroModel class Address(AvroModel): "An Address" street: str street_number: int class User(AvroModel): "User with multiple Address" name: str age: int addresses: typing.Dict[str, Address] # PREVIOUS User.avro_schema() ``` -------------------------------- ### Time example Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/logical_types.md Example demonstrating the use of 'time-millis' and 'time-micros' logical types with Python's datetime.time and types.TimeMicro. ```python import datetime import dataclasses import typing from dataclasses_avroschema import AvroModel, TimeMicro a_datetime = datetime.datetime(2019, 10, 12, 17, 57, 42) @dataclasses.dataclass class TimeLogicalTypes(AvroModel): """Time logical types""" birthday_time: datetime.time meeting_time: typing.Optional[datetime.time] = None release_time: datetime.time = a_datetime.time() release_time_micro: TimeMicro = a_datetime.time() TimeLogicalTypes.avro_schema() { "type": "record", "name": "TimeLogicalTypes", "fields": [ { "name": "birthday_time", "type": {"type": "int", "logicalType": "time-millis"} }, { "name": "meeting_time", "type": ["null", {"type": "int", "logicalType": "time-millis"}], "default": null }, { "name": "release_time", "type": {"type": "int", "logicalType": "time-millis"}, "default": 64662000 }, { "name": "release_time_micro", "type": {"type": "long", "logicalType": "time-micros"}, "default": 64662000000 } ], "doc": "Time logical types" } ``` -------------------------------- ### Examples with null Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/primitive_types.md Demonstrates how to use nullable fields (Optional) with dataclasses-avroschema. ```python import dataclasses from typing import Optional from dataclasses_avroschema import AvroModel, types @dataclasses.dataclass class User(AvroModel): """An User""" name: Optional[str] = None age: Optional[int] = None height: Optional[types.Float32] = None weight: Optional[types.Int32] = None is_student: Optional[bool] = None money_available: Optional[float] = None encoded: Optional[bytes] = None User.avro_schema() ``` -------------------------------- ### Deserialization Examples Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/README.md Examples demonstrating deserialization of Avro binary data back into dataclass instances or Python dictionaries. ```python import typing import dataclasses from dataclasses_avroschema import AvroModel @dataclasses.dataclass class Address(AvroModel): """An Address""" street: str street_number: int @dataclasses.dataclass class User(AvroModel): """User with multiple Address""" name: str age: int addresses: typing.List[Address] avro_binary = b"\x08john(\x02\x08test\x14\x00" avro_json_binary = b'{"name": "john", "age": 20, "addresses": [{"street": "test", "street_number": 10}]}' # return a new class instance!! assert User.deserialize(avro_binary) == User( name='john', age=20, addresses=[Address(street='test', street_number=10)] ) ``` ```python # return a python dict assert User.deserialize(avro_binary, create_instance=False) == { "name": "john", "age": 20, "addresses": [ {"street": "test", "street_number": 10} ] } ``` ```python # return a new class instance!! assert User.deserialize(avro_json_binary, serialization_type="avro-json") == User( name='john', age=20, addresses=[Address(street='test', street_number=10)] ) ``` ```python # return a python dict assert User.deserialize( avro_json_binary, serialization_type="avro-json", create_instance=False ) == {"name": "john", "age": 20, "addresses": [{"street": "test", "street_number": 10}]} ``` -------------------------------- ### Decimal example Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/logical_types.md Example demonstrating the usage of the `condecimal` type for monetary values, including default values and optional fields, and its corresponding Avro schema generation. ```python import decimal import dataclasses import typing from dataclasses_avroschema import AvroModel, types @dataclasses.dataclass class DecimalLogicalTypes(AvroModel): """Decimal logical types""" money: types.condecimal(max_digits=3, decimal_places=2) decimal_with_default: types.condecimal(max_digits=3, decimal_places=2) = decimal.Decimal('3.14') optional_decimal: typing.Optional[types.condecimal(max_digits=3, decimal_places=2)] = None DecimalLogicalTypes.avro_schema() { "type": "record", "name": "DecimalLogicalTypes", "fields": [ { "name": "money", "type": { "type": "bytes", "logicalType": "decimal", "precision": 3, "scale": 2 } }, { "name": "decimal_with_default", "type": { "type": "bytes", "logicalType": "decimal", "precision": 3, "scale": 2 }, "default": "\u013a" }, { "name": "optional_decimal", "type": [ "null", { "type": "bytes", "logicalType": "decimal", "precision": 3, "scale": 2 } ] "default": "null" } ], "doc": "Decimal logical types" } ``` -------------------------------- ### Schema example Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/pydantic.md Example of defining an Avro schema with pydantic types and generating Python code. ```python from dataclasses_avroschema import ModelGenerator, ModelType model_generator = ModelGenerator() schema = { "type": "record", "name": "Infrastructure", "fields": [ {"name": "email", "type": {"type": "string", "pydantic-class": "EmailStr"}}, {"name": "kafka_url", "type": {"type": "string", "pydantic-class": "KafkaDsn"}}, {"name": "total_nodes", "type": {"type": "long", "pydantic-class": "PositiveInt"}}, {"name": "event_id", "type": {"type": "string", "logicalType": "uuid", "pydantic-class": "UUID1"}}, {"name": "landing_zone_nodes", "type": {"type": "array", "items": {"type": "long", "pydantic-class": "PositiveInt"}, "name": "landing_zone_node"}}, {"name": "total_nodes_in_aws", "type": {"type": "long", "pydantic-class": "PositiveInt"}, "default": 10}, {"name": "optional_kafka_url", "type": ["null", {"type": "string", "pydantic-class": "KafkaDsn"}], "default": None} ] } result = model_generator.render(schema=schema, model_type=ModelType.AVRODANTIC.value) with open("models.py", mode="+w") as f: f.write(result) ``` -------------------------------- ### Produce event with metadata Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/good_practices.md Example of producing an event with metadata, including content-type and schema-id, using AIOKafkaProducer. ```python import asyncio import dataclasses from dataclasses_avroschema import AvroModel from aiokafka import AIOKafkaProducer @dataclasses.dataclass class User(AvroModel): """My User Class""" name: str age: int has_pets: bool = False money: float = 100.3 class Meta: schema_id = "https://my-schema-server/users/schema.avsc" # or in a Confluent way: https://my-schema-server/schemas/ids/{int: id} async def produce(): # Naive example of producing an event producer = AIOKafkaProducer(bootstrap_servers='localhost:9092') await producer.start() user = User("Bond", age="50") # create the event event = user.serialize() headers = [ ("content-type": b"avro"), ("schema-id": User.Meta.schema_id.encode()), ] await producer.send_and_wait("my_topic", value=event, headers=headers) await producer.stop() if __name__ == "__main__": asyncio.run(produce) ``` -------------------------------- ### Array example Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/complex_types.md Example of defining a dataclass with list fields and generating its Avro schema. ```python import dataclasses import typing from dataclasses_avroschema import AvroModel @dataclasses.dataclass class UserAdvance(AvroModel): """User advanced""" pets: typing.List[str] cars: typing.List[str] = None favourites_numbers: typing.List[int] = dataclasses.field(default_factory=lambda: [7, 13]) UserAdvance.avro_schema() '{ "type": "record", "name": "UserAdvance", "fields": [ { "name": "pets", "type": { "type": "array", "items": "string", "name": "pet" } }, { "name": "cars", "type": { "type": "array", "items": "string", "name": "car" }, "default": [] }, { "name": "favourites_numbers", "type": { "type": "array", "items": "long", "name": "favourites_number" }, "default": [7, 13] } ], "doc": "User advanced" }' ``` -------------------------------- ### Union encoding with avro-json example Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/serialization.md Demonstrates how unions are encoded with avro-json, including the addition of type information for deserialization and an example of handling malformed data. ```python import typing import dataclasses import datetime import uuid from dataclasses_avroschema import AvroModel @dataclasses.dataclass class UnionSchema(AvroModel): """Some Unions""" first_union: typing.Union[str, int] logical_union: typing.Union[datetime.datetime, datetime.date, uuid.uuid4] my_union = UnionSchema(first_union=10, logical_union=datetime.datetime.now()) event = my_union.serialize(serialization_type="avro-json") print(event) # long is added to each field >>> b'{"first_union": {"long": 10}, "logical_union": {"long": 1647971584847}}' my_union.deserialize(event, serialization_type="avro-json") # >>> UnionSchema(first_union=10, logical_union=datetime.datetime(2022, 3, 22, 17, 53, 4, 847000, tzinfo=datetime.timezone.utc)) # bad data event_2 = b'{"first_union": 10, "logical_union": {"long": 1647971584847}}' my_union.deserialize(event_2, serialization_type="avro-json") File ~/Projects/dataclasses-avroschema/.venv/lib/python3.10/site-packages/fastavro/io/json_decoder.py:213, in AvroJSONDecoder.read_index(self) 211 label = "null" 212 else: --> 213 label, data = self._current[self._key].popitem() 214 self._current[self._key] = data 215 # TODO: Do we need to do this? AttributeError: 'int' object has no attribute 'popitem' ``` -------------------------------- ### Map example Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/complex_types.md Example of defining a dataclass with dictionary fields and generating its Avro schema. ```python import dataclasses import typing from dataclasses_avroschema import AvroModel @dataclasses.dataclass class UserAdvance(AvroModel): """User advanced""" accounts_money: typing.Dict[str, float] cars_brand_total: typing.Dict[str, int] = None family_ages: typing.Dict[str, int] = dataclasses.field(default_factory=lambda: {"father": 50}) UserAdvance.avro_schema() '{ "type": "record", "name": "UserAdvance", "fields": [ { "name": "accounts_money", "type": { "type": "map", "values": "float", "name": "accounts_money" } }, { "name": "cars_brand_total", "type": { "type": "map", "values": "long", "name": "cars_brand_total" }, "default": {} }, { "name": "family_ages", "type": { "type": "map", "values": "long", "name": "family_age" }, "default": {"father": 50} } ], "doc": "User advanced" }' ``` -------------------------------- ### Examples with default values Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/primitive_types.md Demonstrates how to set default values for fields in a dataclass-avroschema model. ```python import dataclasses from dataclasses_avroschema import AvroModel, types @dataclasses.dataclass class User(AvroModel): """An User""" name: str = 'Juan' age: int = 20 height: types.Float32 = 165.3 weight: types.Int32 = 72 is_student: bool = True money_available: float = 100.2 encoded: bytes = b"hi" User.avro_schema() ``` -------------------------------- ### Factory and Fixtures Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/README.md Example of using the fake() method to generate instances of dataclasses. ```python import typing import dataclasses from dataclasses_avroschema import AvroModel @dataclasses.dataclass class Address(AvroModel): """An Address""" street: str street_number: int @dataclasses.dataclass class User(AvroModel): """User with multiple Address""" name: str age: int addresses: typing.List[Address] Address.fake() # >>>> Address(street='PxZJILDRgbXyhWrrPWxQ', street_number=2067) User.fake() # >>>> User(name='VGSBbOGfSGjkMDnefHIZ', age=8974, addresses=[Address(street='vNpPYgesiHUwwzGcmMiS', street_number=4790)]) ``` -------------------------------- ### An User has one Address example Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/schema_relationships.md Demonstrates a OneToOne relationship where a User has a single Address object. ```python import dataclasses from dataclasses_avroschema import AvroModel @dataclasses.dataclass class Address(AvroModel): "An Address" street: str street_number: int class User(AvroModel): "An User with Address" name: str age: int address: Address User.avro_schema() '{...}' ``` -------------------------------- ### An User has multiple Address example Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/schema_relationships.md Shows a OneToMany relationship where a User can have a list of Address objects. ```python import typing import dataclasses from dataclasses_avroschema import AvroModel @dataclasses.dataclass class Address(AvroModel): "An Address" street: str street_number: int @dataclasses.dataclass class User(AvroModel): "User with multiple Address" name: str age: int addresses: typing.List[Address] User.avro_schema() '{...}' ``` -------------------------------- ### Examples with null schema Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/primitive_types.md The Avro schema generated for a User dataclass with nullable fields. ```json { "type": "record", "name": "User", "fields": [ { "name": "name", "type": ["null", "string"], "default": null }, { "name": "age", "type": ["null", "long"], "default": null }, { "name": "height", "type": ["null", "float"], "default": null }, { "name": "weight", "type": ["null", "int"], "default": null }, { "name": "is_student", "type": ["null", "boolean"], "default": null }, { "name": "money_available", "type": ["null", "double"], "default": null }, { "name": "encoded", "type": ["null", "bytes"], "default": null } ], "doc": "An User" } ``` -------------------------------- ### Examples with default values schema Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/primitive_types.md The Avro schema generated for a User dataclass with default values. ```json { "type": "record", "name": "User", "fields": [ { "name": "name", "type": "string", "default": "Juan" }, { "name": "age", "type": "long", "default": 20 }, { "name": "height", "type": ["null", "float"], "default": 165.3 }, { "name": "weight", "type": ["null", "int"], "default": 72 }, { "name": "is_student", "type": "boolean", "default": true }, { "name": "money_available", "type": "double", "default": 100.2 }, { "name": "encoded", "type": "bytes", "default": "hi" } ], "doc": "An User" } ``` -------------------------------- ### Basic usage Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/records.md Example of a basic User record with AvroModel, including namespace and aliases in Meta class. ```python import dataclasses from dataclasses_avroschema import AvroModel @dataclasses.dataclass class User(AvroModel): """My User Class""" name: str age: int has_pets: bool = False money: float = 100.3 class Meta: namespace = "test.com.ar/user/v1" aliases = ["User", "My favorite User"] User.avro_schema() """{ "type": "record", "name": "User", "fields": [ {"name": "name", "type": "string"}, {"name": "age", "type": "long"}, {"name": "has_pets", "type": "boolean", "default": false}, {"name": "money", "type": "double", "default": 100.3} ], "doc": "My User Class", "namespace": "test.com.ar/user/v1", "aliases": ["User", "My favorite User"] }""" ``` -------------------------------- ### Extra Avro attributes in older versions Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/migration_guide.md Demonstrates how extra Avro attributes like 'namespace' and 'aliases' were handled before version 0.14.0, using a separate method. ```python import dataclasses import typing from dataclasses_avroschema import AvroModel class User: "My User Class" name: str age: int has_pets: bool = False money: float = 100.3 def extra_avro_attributes() -> typing.Dict[str, typing.Any]: return { "namespace": "test.com.ar/user/v1", "aliases": ["User", "My favorite User"] } SchemaGenerator(User, include_schema_doc=False).avro_schema() ``` -------------------------------- ### Malformed schema example Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/model_generator.md Example of a schema with a field name that is the same as the record name. ```python { "type": "record", "name": "User", "fields": [ { "name": "name", "type": "string" }, { "name": "age", "type": "long" }, { "name": "Address", # The field name is the same as the record name "type": [ "null", { "type": "record", "name": "Address", "fields": [ { "name": "name", "type": "string" } ] }, ], "default": None, } ] } ``` -------------------------------- ### Enum example Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/complex_types.md Example demonstrating the use of enums in dataclasses-avroschema for Python versions >= 3.11. ```python import enum import dataclasses from dataclasses_avroschema import AvroModel class FavoriteColor(str, enum.Enum): BLUE = "Blue" YELLOW = "Yellow" GREEN = "Green" @enum.nonmember class Meta: doc = "A favorite color" namespace = "some.name.space" aliases = ["Color", "My favorite color"] @dataclasses.dataclass class User(AvroModel): "An User" favorite_color: FavoriteColor = FavoriteColor.BLUE User.avro_schema() '{ "type": "record", "name": "User", "fields": [ { "name": "favorite_color", "type": { "type": "enum", "name": "FavoriteColor", "symbols": [ "Blue", "Yellow", "Green" ], "doc": "A favorite color", "namespace": "some.name.space", "aliases": ["Color", "My favorite color"] }, "default": "Blue" } ], "doc": "An User" }' ``` -------------------------------- ### Enum example Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/complex_types.md Example demonstrating the use of enums in dataclasses-avroschema for Python versions <= 3.10. ```python import enum import dataclasses from dataclasses_avroschema import AvroModel class FavoriteColor(enum.Enum): BLUE = "Blue" YELLOW = "Yellow" GREEN = "Green" class Meta: doc = "A favorite color" namespace = "some.name.space" aliases = ["Color", "My favorite color"] @dataclasses.dataclass class User(AvroModel): "An User" favorite_color: FavoriteColor = FavoriteColor.BLUE User.avro_schema() '{ "type": "record", "name": "User", "fields": [ { "name": "favorite_color", "type": { "type": "enum", "name": "FavoriteColor", "symbols": [ "Blue", "Yellow", "Green" ], "doc": "A favorite color", "namespace": "some.name.space", "aliases": ["Color", "My favorite color"] }, "default": "Blue" } ], "doc": "An User" }' ``` -------------------------------- ### Validation example Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/records.md Example demonstrating the `validate` method for checking if an instance matches its schema. ```python from dataclasses import dataclass from dataclasses_avroschema import AvroModel @dataclass class User(AvroModel): name: str age: int has_pets: bool money: float encoded: bytes # this creates a proper instance user_instance = User( name="a name", age=10, has_pets=True, money=0, encoded=b'hi', ) assert user_instance.validate() ``` -------------------------------- ### Example with CAPITALCASE (Python <= 3.10) Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/case.md Demonstrates generating an Avro schema with CAPITALCASE convention for Python versions up to 3.10. ```python import typing import dataclasses import enum from dataclasses_avroschema import AvroModel, case, types class FavoriteColor(enum.Enum): BLUE = "BLUE" YELLOW = "YELLOW" GREEN = "GREEN" @dataclasses.dataclass class UserAdvance(AvroModel): name: str age: int pets: typing.List[str] accounts: typing.Dict[str, int] favorite_colors: FavoriteColor has_car: bool = False country: str = "Argentina" address: str = None md5: types.Fixed = types.Fixed(16) class Meta: schema_doc = False UserAdvance.avro_schema(case_type=case.CAPITALCASE) ``` ```json { "type": "record", "name": "UserAdvance", "fields": [ {"name": "Name", "type": "string"}, {"name": "Age", "type": "long"}, {"name": "Pets", "type": { "type": "array", "items": "string", "name": "Pet" } }, {"name": "Accounts", "type": { "type": "map", "values": "long", "name": "Account" } }, {"name": "Has_car", "type": "boolean", "default": false}, {"name": "Favorite_colors", "type": { "type": "enum", "name": "FavoriteColor", "symbols": ["BLUE", "YELLOW", "GREEN"] } }, {"name": "Country", "type": "string", "default": "Argentina"}, {"name": "Address", "type": ["null", "string"], "default": null}, {"name": "Md5", "type": {"type": "fixed", "name": "Md5", "size": 16}} ] } ``` -------------------------------- ### fake instance creation Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/pydantic.md Demonstrates how to create fake instances of Pydantic models using the `fake()` class method, which generates random data for fields. ```python import typing import datetime from pydantic import Field from dataclasses_avroschema.pydantic import AvroBaseModel class User(AvroBaseModel): name: str age: int birthday: datetime.date pets: typing.List[str] = Field(default_factory=lambda: ["dog", "cat"]) accounts: typing.Dict[str, int] = Field(default_factory=lambda: {"key": 1}) has_car: bool = False print(User.fake()) # >>> User(name='qWTLkqcIVmSBxpWMpFyR', age=2608, birthday=datetime.date(1982, 3, 30), pets=['wqoEXcJRYjcnJmnIvtiI'], accounts={'JueNdHdzIhHJLc': 779}, has_car=True) ``` -------------------------------- ### Fixed example Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/complex_types.md Example demonstrating the use of the confixed type function with size, namespace, and aliases arguments. ```python import typing import dataclasses from dataclasses_avroschema import AvroModel, types @dataclasses.dataclass class UserAdvance(AvroModel): md5: types.confixed(size=16, namespace='md5', aliases=["md5", "hash"]) name: types.confixed(size=16) = b"u00ffffffffffffx" UnionSchema.avro_schema() { 'type': 'record', 'name': 'UserAdvance', 'fields': [ {'name': 'md5', 'type': {'type': 'fixed', 'name': 'md5', 'size': 16,'namespace': 'md5', 'aliases': ['md5', 'hash']}}, {"name": "name", "type": {"type": "fixed", "name": "name", "size": 16}, "default": "u00ffffffffffffx"} ], 'doc': 'UserAdvance(name: str, md5: dataclasses_avroschema.types.Fixed = 16)'} ``` -------------------------------- ### UUID example Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/logical_types.md Example showcasing the use of the UUID logical type for string fields representing UUIDs. ```python import uuid import dataclasses import typing from dataclasses_avroschema import AvroModel @dataclasses.dataclass class UUIDLogicalTypes(AvroModel): """UUID logical types""" uuid_1: uuid.uuid4 uuid_2: typing.Optional[uuid.uuid4] = None event_uuid: uuid.uuid4 = uuid.uuid4() UUIDLogicalTypes.avro_schema() '{ "type": "record", "name": "UUIDLogicalTypes", "fields": [ { "name": "uuid_1", "type": {"type": "string", "logicalType": "uuid"} }, { "name": "uuid_2", "type": ["null", {"type": "string", "logicalType": "uuid"}], "default": null }, { "name": "event_uuid", "type": {"type": "string", "logicalType": "uuid"}, "default": "ad0677ab-bd1c-4383-9d45-e46c56bcc5c9" } ], "doc": "UUID logical types" }' ``` -------------------------------- ### OneToMany with Map example Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/schema_relationships.md Demonstrates a OneToMany relationship using a dictionary (Avro map) where a User can have multiple Addresses keyed by a string. ```python import typing import dataclasses from dataclasses_avroschema import AvroModel @dataclasses.dataclass class Address(AvroModel): "An Address" street: str street_number: int @dataclasses.dataclass class User(AvroModel): "User with multiple Address" name: str age: int addresses: typing.Dict[str, Address] User.avro_schema() '{...}' ``` -------------------------------- ### Example with CAPITALCASE (Python >= 3.11) Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/case.md Demonstrates generating an Avro schema with CAPITALCASE convention for Python versions 3.11 and above, including a new enum definition. ```python import typing import dataclasses import enum from dataclasses_avroschema import AvroModel, case, types # New enum!! class FavoriteColor(str, enum.Enum): BLUE = "BLUE" YELLOW = "YELLOW" GREEN = "GREEN" @dataclasses.dataclass class UserAdvance(AvroModel): name: str age: int pets: typing.List[str] accounts: typing.Dict[str, int] favorite_colors: FavoriteColor has_car: bool = False country: str = "Argentina" address: str = None md5: types.Fixed = types.Fixed(16) class Meta: schema_doc = False UserAdvance.avro_schema(case_type=case.CAPITALCASE) ``` ```json { "type": "record", "name": "UserAdvance", "fields": [ {"name": "Name", "type": "string"}, {"name": "Age", "type": "long"}, {"name": "Pets", "type": { "type": "array", "items": "string", "name": "Pet" } }, {"name": "Accounts", "type": { "type": "map", "values": "long", "name": "Account" } }, {"name": "Has_car", "type": "boolean", "default": false}, {"name": "Favorite_colors", "type": { "type": "enum", "name": "FavoriteColor", "symbols": ["BLUE", "YELLOW", "GREEN"] } }, {"name": "Country", "type": "string", "default": "Argentina"}, {"name": "Address", "type": ["null", "string"], "default": null}, {"name": "Md5", "type": {"type": "fixed", "name": "Md5", "size": 16}} ] } ``` -------------------------------- ### Repeated types Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/schema_relationships.md Example demonstrating how to use 'namespace' to avoid name collisions when a class is related more than once with another class. ```python from dataclasses import dataclass from datetime import datetime import json from dataclasses_avroschema import AvroModel @dataclass class Location(AvroModel): latitude: float longitude: float class Meta: namespace = "types.location_type" # Good practise to use `namespaces` @dataclass class Trip(AvroModel): start_time: datetime start_location: Location # first relationship finish_time: datetime finish_location: Location # second relationship Trip.avro_schema() ``` ```json { "type": "record", "name": "Trip", "fields": [ { "name": "start_time", "type": {"type": "long", "logicalType": "timestamp-millis"} }, { "name": "start_location", "type": {"type": "record", "name": "Location", "fields": [ {"name": "latitude", "type": "double"}, {"name": "longitude", "type": "double"} ], "doc": "Location(latitude: float, longitude: float)", "namespace": "types.location_type"}}, { "name": "finish_time", "type": {"type": "long", "logicalType": "timestamp-millis"} }, { "name": "finish_location", "type": "types.location_type.Location" // using the namespace and the Location type } ], "doc": "Trip(start_time: datetime.datetime, start_location: __main__.Location, finish_time: datetime.datetime, finish_location: __main__.Location)" } ``` -------------------------------- ### Include schema_id in Meta Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/good_practices.md Example demonstrating how to include a schema_id in the Meta class of an AvroModel, which can be used to reference schemas in a schema server. ```python import dataclasses from dataclasses_avroschema import AvroModel @dataclasses.dataclass class User(AvroModel): """My User Class""" name: str age: int has_pets: bool = False money: float = 100.3 class Meta: schema_id = "https://my-schema-server/users/schema.avsc" # or in a Concluent way: https://my-schema-server/schemas/ids/{int: id} ``` -------------------------------- ### Basic usage Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/factories_and_fixtures.md Generates fake instances of dataclasses-avroschema models using the `fake` method. ```python import typing import dataclasses from dataclasses_avroschema import AvroModel @dataclasses.dataclass class Address(AvroModel): """An Address""" street: str street_number: int @dataclasses.dataclass class User(AvroModel): """User with multiple Address""" name: str age: int addresses: typing.List[Address] Address.fake() # >>>> Address(street='PxZJILDRgbXyhWrrPWxQ', street_number=2067) User.fake() # >>>> User(name='VGSBbOGfSGjkMDnefHIZ', age=8974, addresses=[Address(street='vNpPYgesiHUwwzGcmMiS', street_number=4790)]) ``` -------------------------------- ### Getting dict and json Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/pydantic.md Demonstrates how to get dictionary and JSON representations of a Pydantic model instance using dataclasses-avroschema methods, which are equivalent to Pydantic's `model_dump` and `model_dump_json`. ```python user = UserAdvance(name="bond", age=50) # to_json from dataclasses-avroschema is the same that json from pydantic assert user.to_json(separators=((",",":",))) == user.model_dump_json() # to_dict from dataclasses-avroschema is the same that dict from pydantic assert user.to_dict() == user.model_dump() ``` -------------------------------- ### Avrodantic models Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/model_generator.md Example of rendering Avrodantic models from a schema. ```python from dataclasses_avroschema import ModelGenerator model_generator = ModelGenerator() result = model_generator.render(schema=schema, model_type=ModelType.AVRODANTIC.value) # save the result in a file with open("models.py", mode="+w") as f: f.write(result) # models.py from dataclasses_avroschema.pydantic import AvroBaseModel from pydantic import Field from pydantic import condecimal import typing class User(AvroBaseModel): name: str age: int money: condecimal(max_digits=10, decimal_places=3) friend: typing.Optional[typing.Type["User"]] = None relatives: typing.List[typing.Type["User"]] = Field(default_factory=list) teammates: typing.Dict[str, typing.Type["User"]] = Field(default_factory=dict) ``` -------------------------------- ### Extra Avro attributes in new versions (>= 0.14.0) Source: https://github.com/marcosschroh/dataclasses-avroschema/blob/master/docs/migration_guide.md Shows the modern approach to defining extra Avro attributes using a Meta class within the AvroModel dataclass. ```python # Now is perform using a Meta class @dataclasses.dataclass class User(AvroModel): "My User Class" name: str age: int has_pets: bool = False money: float = 100.3 class Meta: schema_doc = False namespace = "test.com.ar/user/v1" aliases = ["User", "My favorite User"] ```