Dataclasses Avro Schema

https://marcosschroh.github.io/dataclasses-avroschema/
Admin
Dataclasses Avro Schema is a Python library that generates Apache Avro schemas from Python...
Tokens:41,211
Snippets:281
Trust Score:9.7
Update:6 months ago
Show doc for...
Tokens:
Raw
### Install dataclasses-avroschema with extras

Source: https://marcosschroh.github.io/dataclasses-avroschema

Install the library with specific extras like pydantic, faust-streaming, faker, or cli using pip or poetry.

```bash
pip install dataclasses-avroschema[pydantic]
```

```bash
poetry add dataclasses-avroschema --extras "pydantic"
```

```bash
pip install 'dataclasses-avroschema[faust]'
```

```bash
poetry add dataclasses-avroschema --extras "faust "
```

```bash
pip install 'dataclasses-avroschema[faker]'
```

```bash
poetry add dataclasses-avroschema --extras "faker"
```

```bash
pip install 'dataclasses-avroschema[cli]'
```

```bash
poetry add dataclasses-avroschema --with cli
```

```bash
pip install dataclasses-avroschema[faust,pydantic,faker,cli]
```

```bash
poetry add dataclasses-avroschema --extras "pydantic faust faker cli"
```

--------------------------------

### Redis Streams Initialization Example

Source: https://marcosschroh.github.io/dataclasses-avroschema/examples

Initializes a Redis database connection, creates a stream, and sets up a consumer group for the stream. This is a prerequisite for Redis Streams producer and consumer examples.

```python
from walrus import Database

def main():
    db = Database()
    stream_name = "my-stream"
    db.Stream(stream_name)  # Create a new stream instance

    # create the consumer group
    consumer_group = db.consumer_group("my-consumer-group-1", [stream_name])
    consumer_group.create()  # Create the consumer group.
    consumer_group.set_id("$")

    produce(consumer_group)
    consume(consumer_group)
```

--------------------------------

### Produce event with metadata

Source: https://marcosschroh.github.io/dataclasses-avroschema/good_practices

Example of serializing an AvroModel and sending it to Kafka with content-type and schema-id headers.

```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)
```

--------------------------------

### Kafka Consumer Example

Source: https://marcosschroh.github.io/dataclasses-avroschema/examples

Consumes Avro-serialized messages from a Kafka topic and deserializes them into UserModelV2 objects. Requires a running Kafka instance.

```python
from kafka import KafkaConsumer

def consume():
    consumer = KafkaConsumer("my_topic", bootstrap_servers="localhost:9092", group_id="my-group")

    for msg in consumer:
        print(f"Message received: {msg.value} at {msg.timestamp}")

        # create an instance of User v2
        user = UserModelV2.deserialize(msg.value)
        print(f"Message deserialized: {user}")

    print("Stoping consumer...")
```

--------------------------------

### Schema Evolution Example

Source: https://marcosschroh.github.io/dataclasses-avroschema/examples

Defines a UserModel dataclass intended for use with Avro schemas, illustrating the base structure for schema evolution examples. This snippet focuses on the data model definition itself.

```python
import enum
import random
from dataclasses import dataclass
from time import sleep
from typing import Optional

from kafka import KafkaConsumer, KafkaProducer

from dataclasses_avroschema import AvroModel


class FavoriteColor(enum.Enum):
    BLUE = "BLUE"
    YELLOW = "YELLOW"
    GREEN = "GREEN"


@dataclass
class UserModel(AvroModel):
    "An User"

    name: str
    age: int
    favorite_colors: FavoriteColor = FavoriteColor.BLUE
    country: str = "Argentina"
    address: Optional[str] = None

```

--------------------------------

### Kafka Producer Example

Source: https://marcosschroh.github.io/dataclasses-avroschema/examples

Produces Avro-serialized UserModel instances to a Kafka topic. Includes random data generation and a delay between messages. Requires a running Kafka instance.

```python
import random
from time import sleep
from kafka import KafkaProducer

def send(total_events=2):
    producer = KafkaProducer(bootstrap_servers="localhost:9092")

    for event_number in range(1, total_events + 1):
        # Produce message
        print(f"Sending event number {event_number}")

        # create an instance of User v1
        user = UserModel(
            name=random.choice(
                [
                    "Juan",
                    "Peter",
                    "Michael",
                    "Moby",
                    "Kim",
                ]
            ),
            age=random.randint(1, 50),
        )

        print(user)

        # create the message
        message = user.serialize()

        producer.send("my_topic", message)
        # sleep for 2 seconds
        sleep(1)

    print("Stoping producer...")
```

--------------------------------

### Define User Avro Schema

Source: https://marcosschroh.github.io/dataclasses-avroschema/migration_guide

Example of a User dataclass with nested Address type, demonstrating basic Avro schema generation.

```python
import dataclasses


@dataclasses.dataclass
class Address:
    street: str
    street_number: long


@dataclasses.dataclass
class User:
    name: str
    age: long
    addresses: dict[str, Address]

User.avro_schema()
```

--------------------------------

### Example of Invalid Data

Source: https://marcosschroh.github.io/dataclasses-avroschema/faust_records

Demonstrates behavior when validation is not explicitly enabled.

```python
import typing

from dataclasses_avroschema.faust import AvroRecord


class UserAdvance(AvroRecord):
    name: str
    age: int
    pets: typing.List[str] = fields.StringField(required=False, default=['dog', 'cat'])
    accounts: typing.Dict[str, int] = fields.IntegerField(required=False, default={"key": 1})
    has_car: bool = fields.BooleanField(required=False, default=False)
    favorite_colors: typing.Tuple[str] = fields.StringField(required=False, default=("BLUE", "YELLOW", "GREEN"))
    country: str = fields.StringField(required=False, default="Argentina")
    address: typing.Optional[str] = None

    class Meta:
        schema_doc = False

UserAdvance(name="marcos", age="juan")  # WRONG data
```

--------------------------------

### Nested Record Schema Generation

Source: https://marcosschroh.github.io/dataclasses-avroschema/migration_guide

Example of nested record schema generation using AvroModel.

```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()
{
  "type": "record",
  "name": "User",
  "fields": [
    {"name": "name", "type": "string"},
    {"name": "age", "type": "long"},
    {"name": "addresses", "type": {
        "type": "map",
        "values": {
          "type": "record",
          "name": "address_record",
          "fields": [
            {"name": "street", "type": "string"},
            {"name": "street_number", "type": "long"}
          ],
          "doc": "An Address"
        },
        "name": "address"
      }
    }
  ],
  "doc": "User with multiple Address"
}
```

--------------------------------

### Avro Record Schema Example

Source: https://marcosschroh.github.io/dataclasses-avroschema/avro_schema

Defines a User record with basic fields like name, age, has_pets, and money using Avro's JSON schema format.

```json
{
    "type": "record",
    "name": "User",
    "fields" : [
        {"name": "name", "type": "string"},
        {"name": "age", "type": "int"},
        {"name": "has_pets", "type": "boolean"},
        {"name": "money", "type": "float"}
    ]
}

```

--------------------------------

### User Schema Generation (Versions < 0.14.0)

Source: https://marcosschroh.github.io/dataclasses-avroschema/migration_guide

Example of generating an Avro schema using SchemaGenerator for versions prior to 0.14.0. This approach is now deprecated.

```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()
```

--------------------------------

### Generated Python Module from Multiple Avro Schemas

Source: https://marcosschroh.github.io/dataclasses-avroschema/model_generator

Example of a Python module containing multiple dataclasses generated from different Avro schemas. Demonstrates default values and type mappings.

```python
# models.py
from dataclasses_avroschema import AvroModel
from dataclasses_avroschema import types
import dataclasses


@dataclasses.dataclass
class User(AvroModel):
    age: types.Int32
    name: str = "marcos"


@dataclasses.dataclass
class Address(AvroModel):
    street: str
    street_number: int

```

--------------------------------

### Redis Streams Producer Example

Source: https://marcosschroh.github.io/dataclasses-avroschema/examples

Produces Avro-serialized UserModel instances to a Redis stream using the walrus driver. Generates 10 messages with random user data. Assumes a Redis instance is running.

```python
import random
from time import sleep

def produce(consumer_group):
    for i in range(10):
        # create an instance of User v1
        user = UserModel(
            name=random.choice(
                [
                    "Juan",
                    "Peter",
                    "Michael",
                    "Moby",
                    "Kim",
                ]
            ),
            age=random.randint(1, 50),
        )

        msgid = consumer_group.my_stream.add({"message": user.serialize()})
        print(f"Producing message {msgid}")

    print("Producer finished....")
    print("#" * 80)
    sleep(2)
```

--------------------------------

### Redis Streams Consumer Example

Source: https://marcosschroh.github.io/dataclasses-avroschema/examples

Reads and processes messages from a Redis stream using the walrus driver. Deserializes messages into UserModel objects. Assumes a Redis instance is running.

```python
def consume(consumer_group):
    # read new messages in the stream

    while True:
        result = consumer_group.my_stream.read(count=1, block=1000)
        # Each record has the followinf format
        # [(b'1598545738231-0', {b'message': b'\x06KimT\x00\x12Argentina\x00\x00'})

        if result:
            message_id, message_content = result[0]

            if message_id:
                value = message_content[b"message"]
                print(f"Processing message {message_id} with value {value}")
                user = UserModel.deserialize(value)
                print(user)
```

--------------------------------

### Generated Python Dataclass from Avro Schema

Source: https://marcosschroh.github.io/dataclasses-avroschema/model_generator

Example of a Python dataclass generated from an Avro schema using ModelGenerator. It includes necessary imports and metadata.

```python
import dataclasses

from dataclasses_avroschema import AvroModel
from dataclasses_avroschema import types


@dataclasses.dataclass
class AvroDeployment(AvroModel):
    image: str
    replicas: types.Int32
    port: types.Int32

    class Meta:
        namespace = "com.kubertenes"

```

--------------------------------

### Generate Avro Schema from Dataclass (Python 3.10)

Source: https://marcosschroh.github.io/dataclasses-avroschema

Define a dataclass inheriting from AvroModel and use User.avro_schema() to generate its Avro schema. This example uses a standard enum.

```python
import enum
import typing
import dataclasses

from dataclasses_avroschema import AvroModel


class FavoriteColor(enum.Enum):
    BLUE = "Blue"
    YELLOW = "Yellow"
    GREEN = "Green"


@dataclasses.dataclass
class User(AvroModel):
    """An User"""
    name: str
    age: int
    pets: typing.List[str]
    accounts: typing.Dict[str, int]
    favorite_color: FavoriteColor
    country: str = "Argentina"
    address: str = None

    class Meta:
        namespace = "User.v1"
        aliases = ["user-v1", "super user"]


User.avro_schema()
```

```json
{
    "type": "record",
    "name": "User",
    "doc": "An User",
    "namespace": "User.v1",
    "aliases": ["user-v1", "super user"],
    "fields": [
        {"name": "name", "type": "string"},
        {"name": "age", "type": "long"},
        {"name": "pets", "type": "array", "items": "string"},
        {"name": "accounts", "type": "map", "values": "long"},
        {"name": "favorite_color", "type": {"type": "enum", "name": "FavoriteColor", "symbols": ["Blue", "Yellow", "Green"]}}
        {"name": "country", "type": "string", "default": "Argentina"},
        {"name": "address", "type": ["null", "string"], "default": null}
    ]
}
```

```python
User.avro_schema_to_python()
```

```json
{
    "type": "record",
    "name": "User",
    "doc": "An User",
    "namespace": "User.v1",
    "aliases": ["user-v1", "super user"],
    "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": "favorite_color", "type": {"type": "enum", "name": "FavoriteColor", "symbols": ["BLUE", "YELLOW", "GREEN"]}},
        {"name": "country", "type": "string", "default": "Argentina"},
        {"name": "address", "type": ["null", "string"], "default": None}
    ],
}
```

--------------------------------

### Generate Avro Schema from Dataclass (Python >= 3.11)

Source: https://marcosschroh.github.io/dataclasses-avroschema

Define a dataclass inheriting from AvroModel and use User.avro_schema() to generate its Avro schema. This example uses an enum that inherits from str.

```python
import enum
import typing
import dataclasses

from dataclasses_avroschema import AvroModel


class FavoriteColor(str, enum.Enum):
    BLUE = "Blue"
    YELLOW = "Yellow"
    GREEN = "Green"


@dataclasses.dataclass
class User(AvroModel):
    """An User"""
    name: str
    age: int
    pets: typing.List[str]
    accounts: typing.Dict[str, int]
    favorite_color: FavoriteColor
    country: str = "Argentina"
    address: str = None

    class Meta:
        namespace = "User.v1"
        aliases = ["user-v1", "super user"]


User.avro_schema()
```

```json
{
    "type": "record",
    "name": "User",
    "doc": "An User",
    "namespace": "User.v1",
    "aliases": ["user-v1", "super user"],
    "fields": [
        {"name": "name", "type": "string"},
        {"name": "age", "type": "long"},
        {"name": "pets", "type": "array", "items": "string"},
        {"name": "accounts", "type": "map", "values": "long"},
        {"name": "favorite_color", "type": {"type": "enum", "name": "FavoriteColor", "symbols": ["Blue", "Yellow", "Green"]}}
        {"name": "country", "type": "string", "default": "Argentina"},
        {"name": "address", "type": ["null", "string"], "default": null}
    ]
}
```

```python
User.avro_schema_to_python()
```

```json
{
    "type": "record",
    "name": "User",
    "doc": "An User",
    "namespace": "User.v1",
    "aliases": ["user-v1", "super user"],
    "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": "favorite_color", "type": {"type": "enum", "name": "FavoriteColor", "symbols": ["BLUE", "YELLOW", "GREEN"]}},
        {"name": "country", "type": "string", "default": "Argentina"},
        {"name": "address", "type": ["null", "string"], "default": None}
    ],
}
```

--------------------------------

### Generate Python Model with Case-Sensitive Enum Symbols

Source: https://marcosschroh.github.io/dataclasses-avroschema/model_generator

This example demonstrates generating a Python model from an Avro schema containing an enum with case-sensitive symbols. It's generally advised to avoid such schemas.

```python
from dataclasses_avroschema import ModelGenerator, ModelType

schema = {
    "type": "record",
    "name": "User",
    "fields": [
        {
            "name": "unit_multi_player",
            "type": {
                "type": "enum",
                "name": "unit_multi_player",
                "symbols": ["Q", "q"],
            },
        }
    ],
}

model_generator = ModelGenerator()
result = model_generator.render(schema=schema, model_type=ModelType.DATACLASS.value)

# save the result in a file
with open("models.py", mode="+w") as f:
    f.write(result)
```

--------------------------------

### Using typing.Literal with Enum for Avro Schema (Python <= 3.10)

Source: https://marcosschroh.github.io/dataclasses-avroschema/fields_specification

Demonstrates using typing.Literal with an enum for a field in Avro schema generation. This example is for Python versions less than or equal to 3.10.

```python
import enum
import typing
from dataclasses import dataclass
from dataclasses_avroschema import AvroModel

class E(enum.Enum):
    ONE = "one"

@dataclass
class T(AvroModel):
    f: typing.Literal[None, 1, "1", True, b"1", E.ONE]

print(T.avro_schema())
```

--------------------------------

### Using typing.Literal with Enum for Avro Schema (Python >= 3.11)

Source: https://marcosschroh.github.io/dataclasses-avroschema/fields_specification

Demonstrates using typing.Literal with an enum for a field in Avro schema generation. This example is for Python versions greater than or equal to 3.11.

```python
import enum
import typing
from dataclasses import dataclass
from dataclasses_avroschema import AvroModel

class E(str, enum.Enum):
    ONE = "one"

@dataclass
class T(AvroModel):
    f: typing.Literal[None, 1, "1", True, b"1", E.ONE]

print(T.avro_schema())
```

--------------------------------

### Implement class inheritance

Source: https://marcosschroh.github.io/dataclasses-avroschema/records

Shows how to inherit from AvroModel classes, noting potential issues with default values in parent classes.

```python
from dataclasses import dataclass

from dataclasses_avroschema import AvroModel


@dataclass
class Parent(AvroModel):
    name: str
    age: int


@dataclass
class Child(Parent):
    has_pets: bool
    money: float
    encoded: bytes


@dataclass
class Child2(Parent, AvroModel):
    has_pets: bool
    money: float
    encoded: bytes

    class Meta:
        schema_doc = False


child_schema = Child.avro_schema_to_python()
child_2_schema = Child2.avro_schema_to_python()


assert child_schema["fields"] == child_2_schema["fields"]
```

--------------------------------

### RabbitMQ Producer and Consumer with Avro

Source: https://marcosschroh.github.io/dataclasses-avroschema/examples

Demonstrates how to define an Avro-compatible dataclass and use it to serialize and deserialize messages sent via RabbitMQ.

```python
import enum
import random
from dataclasses import dataclass
from typing import Optional

import pika

from dataclasses_avroschema import AvroModel


class FavoriteColor(enum.Enum):
    BLUE = "BLUE"
    YELLOW = "YELLOW"
    GREEN = "GREEN"


@dataclass
class UserModel(AvroModel):
    "An User"

    name: str
    age: int
    favorite_colors: FavoriteColor = FavoriteColor.BLUE
    country: str = "Argentina"
    address: Optional[str] = None

    class Meta:
        namespace = "User.v1"
        aliases = ["user-v1", "super user"]


def consume(*, channel, queue_name: str):
    def callback(channel: str, method, properties, body):
        print(f"Message received with body: {body} in channel {method} \n")
        user = UserModel.deserialize(body)
        print(f"Message deserialized: {user} \n")

    channel.basic_consume(on_message_callback=callback, queue=queue_name)
    channel.start_consuming()


def produce(*, channel, queue_name: str, total_events: int = 10):
    for event_number in range(1, total_events + 1):
        # Produce message
        print(f"Sending event number {event_number}")

        user = UserModel(
            name=random.choice(
                [
                    "Juan",
                    "Peter",
                    "Michael",
                    "Moby",
                    "Kim",
                ]
            ),
            age=random.randint(1, 50),
        )

        # create the message
        message = user.serialize()

        # publish
        channel.basic_publish(exchange="", routing_key=queue_name, body=message)


def main():
    rabbitmq_url = "amqp://guest:guest@localhost:5672/%2f"
    queue_name = "test-stream"
    params = pika.URLParameters(rabbitmq_url)

    connection = pika.BlockingConnection(params)
    channel = connection.channel()
    channel.queue_declare(queue=queue_name, durable=True)

    produce(channel=channel, queue_name=queue_name)
    consume(channel=channel, queue_name=queue_name)

    connection.close()
```

--------------------------------

### Generate fake model instances

Source: https://marcosschroh.github.io/dataclasses-avroschema/pydantic

Demonstrates creating randomized model instances using the fake() method.

```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={'JueNdHdzIhHIDsjlHJLc': 779}, has_car=True)
```

--------------------------------

### Parse objects with AvroBaseModel

Source: https://marcosschroh.github.io/dataclasses-avroschema/pydantic

Shows how to use parse_obj to instantiate models from dictionaries.

```python
import typing

from dataclasses_avroschema.pydantic import AvroBaseModel


class Address(AvroBaseModel):
    "An Address"
    street: str
    street_number: int


class User(AvroBaseModel):
    "User with multiple Address"
    name: str
    age: int
    addresses: typing.List[Address]

data_user = {
    "name": "john",
    "age": 20,
    "addresses": [{
        "street": "test",
        "street_number": 10,
        }],
    }

user = User.parse_obj(data=data_user)
assert type(user.addresses[0]) is Address
```

--------------------------------

### Serialize and Deserialize Data

Source: https://marcosschroh.github.io/dataclasses-avroschema/pydantic_v1

Demonstrates converting models to dictionaries or JSON and performing binary serialization and deserialization.

```python
user = UserAdvance(name="bond", age=50)

# to_json from dataclasses-avroschema is the same that json from pydantic
assert user.to_json() == user.json()

# to_dict from dataclasses-avroschema is the same that dict from pydantic
assert user.to_dict() == user.dict()
```

```python
event = user.serialize()
print(event)
# >>> b'\x08bondd\x04\x06dog\x06cat\x00\x02\x06key\x02\x00\x00\x00\x12Argentina\x00'

UserAdvance.deserialize(data=event)
# >>> UserAdvance(name='bond', age=50, pets=['dog', 'cat'], accounts={'key': 1}, has_car=False, favorite_colors=<FavoriteColor.BLUE: 'BLUE'>, country='Argentina', address=None)
```

--------------------------------

### Define Primitive Type as Defined Type

Source: https://marcosschroh.github.io/dataclasses-avroschema/fields_specification

Example of defining an Avro schema using a primitive type with additional properties like unit and default.

```json
{
  "name": "expirience",
  "type": {"type": "int", "unit": "years", "default": 10}
}
```

--------------------------------

### Default typing.Literal to Avro String

Source: https://marcosschroh.github.io/dataclasses-avroschema/complex_types

Without `convert_literal_to_enum = True`, `typing.Literal` string arguments are converted to Avro strings. This example includes an optional field.

```python
import dataclasses
import typing
from dataclasses_avroschema import AvroModel

@dataclasses.dataclass
class MyModel(AvroModel):
    options: typing.Literal["four_4", "five_5"] = "four_4"
    optional_field: typing.Optional[typing.Literal["four_4", "five_5"]] = None

print(MyModel.avro_schema())
```

--------------------------------

### Synchronous Kafka Producer and Consumer with kafka-python

Source: https://marcosschroh.github.io/dataclasses-avroschema/examples

Demonstrates synchronous message production and consumption using kafka-python. Requires Kafka running on localhost:9092 and the 'my_topic' topic. Messages are serialized using AvroModel and deserialized back into UserModel objects.

```python
import enum
import random
from dataclasses import dataclass
from time import sleep
from typing import Optional

from kafka import KafkaConsumer, KafkaProducer

from dataclasses_avroschema import AvroModel


class FavoriteColor(enum.Enum):
    BLUE = "BLUE"
    YELLOW = "YELLOW"
    GREEN = "GREEN"


@dataclass
class UserModel(AvroModel):
    "An User"

    name: str
    age: int
    favorite_colors: FavoriteColor = FavoriteColor.BLUE
    country: str = "Argentina"
    address: Optional[str] = None


def consume():
    consumer = KafkaConsumer("my_topic", bootstrap_servers="localhost:9092", group_id="my-group")

    for msg in consumer:
        print(f"Message received: {msg.value} at {msg.timestamp}")

        user = UserModel.deserialize(msg.value)
        print(f"Message deserialized: {user}")

    print("Stoping consumer...")


def send(total_events=10):
    producer = KafkaProducer(bootstrap_servers="localhost:9092")

    for event_number in range(1, total_events + 1):
        # Produce message
        print(f"Sending event number {event_number}")

        # create an instance of User v1
        user = UserModel(
            name=random.choice(
                [
                    "Juan",
                    "Peter",
                    "Michael",
                    "Moby",
                    "Kim",
                ]
            ),
            age=random.randint(1, 50),
        )

        # create the message
        message = user.serialize()

        producer.send("my_topic", message)
        # sleep for 2 seconds
        sleep(1)

    print("Stoping producer...")


def main():
    send()
    consume()

```

--------------------------------

### Render Avro Enum in Isolation

Source: https://marcosschroh.github.io/dataclasses-avroschema/model_generator

Avro enums can be rendered independently. This example shows how to generate a Python enum class from an Avro enum schema.

```python
from dataclasses_avroschema import ModelGenerator


enum_schema = {
    "type": "enum",
    "name": "Color",
    "symbols": [
        "red",
        "blue",
    ],
    "default": "blue",
}

model_generator = ModelGenerator()
result = model_generator.render(schema=enum_schema)

print(result)
```

--------------------------------

### Generate Fake Data Instances

Source: https://marcosschroh.github.io/dataclasses-avroschema/pydantic_v1

Demonstrates creating randomized instances of Pydantic models using the fake method.

```python
import typing
import datetime
from pydantic.v1 import Field
from dataclasses_avroschema.pydantic.v1 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={'JueNdHdzIhHIDsjlHJLc': 779}, has_car=True)
```

--------------------------------

### Parse Objects with AvroBaseModel

Source: https://marcosschroh.github.io/dataclasses-avroschema/pydantic_v1

Shows how to parse dictionary data into Pydantic models using parse_obj and parse_obj_as.

```python
import typing

from dataclasses_avroschema.pydantic.v1 import AvroBaseModel


class Address(AvroBaseModel):
    "An Address"
    street: str
    street_number: int


class User(AvroBaseModel):
    "User with multiple Address"
    name: str
    age: int
    addresses: typing.List[Address]

data_user = {
    "name": "john",
    "age": 20,
    "addresses": [{
        "street": "test",
        "street_number": 10,
        }],
    }

user = User.parse_obj(data=data_user)
assert type(user.addresses[0]) is Address
```

```python
from typing import List

from pydantic import parse_obj_as

from dataclasses_avroschema.pydantic.v1 import AvroBaseModel


class User(AvroBaseModel):
    "User with multiple Address"
    name: str
    age: int


data = [{"name": "bond", "age": 50}, {"name": "bond2", "age": 60}]
users = parse_obj_as(List[User], data)

users[0].avro_schema()
# '{"type": "record", "name": "User", "fields": [{"name": "name", "type": "string"}, {"name": "age", "type": "long"}], "doc": "User with multiple Address"}'
```

--------------------------------

### Convert typing.Literal to Avro Enum

Source: https://marcosschroh.github.io/dataclasses-avroschema/complex_types

Set `convert_literal_to_enum = True` in the `Meta` class to automatically convert `typing.Literal` string arguments into Avro enums. This example includes an optional field.

```python
import dataclasses
import typing
from dataclasses_avroschema import AvroModel

@dataclasses.dataclass
class MyModel(AvroModel):
    options: typing.Literal["four_4", "five_5"] = "four_4"
    optional_field: typing.Optional[typing.Literal["four_4", "five_5"]] = None

    class Meta:
        convert_literal_to_enum = True

print(MyModel.avro_schema())
```

--------------------------------

### Class Meta Configuration

Source: https://marcosschroh.github.io/dataclasses-avroschema/records

Configuration options for the Meta inner class to customize Avro schema generation.

```APIDOC
## Class Meta Configuration

### Description
The `class Meta` allows customization of the generated Avro schema attributes.

### Configuration Options
- **schema_name** (str): Custom name for the schema.
- **schema_doc** (Union[bool, str]): Whether to include docstrings as schema documentation.
- **namespace** (str): Schema namespace.
- **aliases** (List[str]): Schema aliases.
- **alias_nested_items** (Dict[str, str]): Custom naming for nested items.
- **field_order** (List[str]): Order of fields in the output schema.
- **exclude** (List[str]): Fields to exclude from the schema.
- **convert_literal_to_enum** (bool): Convert Literal strings to enums.
- **dacite_config** (Dict): Custom configuration for dacite parsing.
```

--------------------------------

### Serialize and convert Pydantic models

Source: https://marcosschroh.github.io/dataclasses-avroschema/pydantic

Demonstrates how to convert models to dictionaries or JSON and perform Avro serialization.

```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()
```

```python
event = user.serialize()
print(event)
# >>> b'\x08bondd\x04\x06dog\x06cat\x00\x02\x06key\x02\x00\x00\x00\x12Argentina\x00'

UserAdvance.deserialize(data=event)
# >>> UserAdvance(name='bond', age=50, pets=['dog', 'cat'], accounts={'key': 1}, has_car=False, favorite_colors=<FavoriteColor.BLUE: 'BLUE'>, country='Argentina', address=None)
```

--------------------------------

### Async Kafka Producer and Consumer with aiokafka

Source: https://marcosschroh.github.io/dataclasses-avroschema/examples

Demonstrates asynchronous message production and consumption using aiokafka. Ensure Kafka is running on localhost:9092 and topics 'my_topic' and 'my_other_topic' are available. The consumer will deserialize Avro messages into UserModel objects.

```python
import asyncio
import enum
import random
from dataclasses import dataclass
from typing import Optional

from aiokafka import AIOKafkaConsumer, AIOKafkaProducer

from dataclasses_avroschema import AvroModel


class FavoriteColor(enum.Enum):
    BLUE = "BLUE"
    YELLOW = "YELLOW"
    GREEN = "GREEN"


@dataclass
class UserModel(AvroModel):
    "An User"

    name: str
    age: int
    favorite_colors: FavoriteColor = FavoriteColor.BLUE
    country: str = "Argentina"
    address: Optional[str] = None

    class Meta:
        namespace = "User.v1"
        aliases = ["user-v1", "super user"]


async def consume(loop, total_events=10):
    consumer = AIOKafkaConsumer(
        "my_topic",
        "my_other_topic",
        loop=loop,
        bootstrap_servers="localhost:9092",
        group_id="my-group",
    )
    # Get cluster layout and join group `my-group`
    await consumer.start()
    run_consumer = True

    while run_consumer:
        try:
            # Consume messages
            async for msg in consumer:
                print(f"Message received: {msg.value} at {msg.timestamp}")

                user = UserModel.deserialize(msg.value)
                print(f"Message deserialized: {user}")
        except KeyboardInterrupt:
            # Will leave consumer group; perform autocommit if enabled.
            await consumer.stop()
            print("Stoping consumer...")
            run_consumer = False


async def send(loop, total_events=10):
    producer = AIOKafkaProducer(loop=loop, bootstrap_servers="localhost:9092")
    # Get cluster layout and initial topic/partition leadership information
    await producer.start()

    for event_number in range(1, total_events + 1):
        # Produce message
        print(f"Sending event number {event_number}")

        user = UserModel(
            name=random.choice(
                [
                    "Juan",
                    "Peter",
                    "Michael",
                    "Moby",
                    "Kim",
                ]
            ),
            age=random.randint(1, 50),
        )

        # create the message
        message = user.serialize()

        await producer.send_and_wait("my_topic", message)
        # sleep for 2 seconds
        await asyncio.sleep(2)
    else:
        # Wait for all pending messages to be delivered or expire.
        await producer.stop()
        print("Stoping producer...")


def main():
    loop = asyncio.get_event_loop()
    tasks = asyncio.gather(send(loop), consume(loop))
    loop.run_until_complete(tasks)

```

--------------------------------

### Define AvroModel with Default Values

Source: https://marcosschroh.github.io/dataclasses-avroschema/primitive_types

Illustrates how to assign default values to fields in the dataclass, which are then reflected in the Avro schema.

```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()
```

```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"
}
```

--------------------------------

### Define dataclass with Union types (Python <= 3.9)

Source: https://marcosschroh.github.io/dataclasses-avroschema/complex_types

Use typing.Union to define fields that can hold values of multiple types. This example shows unions for basic types and logical types.

```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]
    second_union: typing.Union[str, int] = "test"
    third_union: typing.Union[int, str] = 'STRING TYPE'

UnionSchema.avro_schema()
```

--------------------------------

### Define Enum with Meta for Avro Schema

Source: https://marcosschroh.github.io/dataclasses-avroschema/complex_types

Use `enum.Enum` with a `Meta` class to define Avro enum schema properties like doc, namespace, and aliases. This example is for Python <= 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()
```

--------------------------------

### Provide Custom Data to the Factory

Source: https://marcosschroh.github.io/dataclasses-avroschema/factories_and_fixtures

Pass keyword arguments to the `fake` method to override default generated values. This allows for more control when creating fake instances.

```python
import typing
import dataclasses

from dataclasses_avroschema import AvroModel


@dataclasses.dataclass
class UserAdvance(AvroModel):
    name: str
    age: int
    pets: typing.List[str]
    accounts: typing.Dict[str, int]
    has_car: bool = False
    country: str = "Argentina"
    address: typing.Optional[str] = None


user = UserAdvance.fake(name="bond", age=50, pets=["dog", "cat"])
assert user.name == "bond"
assert user.age == 50
assert user.pets == ["dog", "cat"]

print(user)
# >>>> UserAdvance(name='bond', age=50, pets=['dog', 'cat'], accounts={'uVITaqdTStKhsdHFqIdM': 4201}, has_car=True, country='cirJWyuMaXoBqEmxbdML', address='qGhXKxAFfIxzAMZdkhrk')

```

--------------------------------

### Define Union Schemas with default_factory

Source: https://marcosschroh.github.io/dataclasses-avroschema/migration_guide

Demonstrates defining Union types in AvroModel with default_factory, ensuring callables return appropriate types.

```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()
```

--------------------------------

### Extra Avro Attributes (Current Method)

Source: https://marcosschroh.github.io/dataclasses-avroschema/migration_guide

Shows how to define extra Avro attributes like 'namespace', 'aliases', and 'schema_doc' using the Meta class within the dataclass.

```python
import dataclasses
import typing

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_doc = False
        namespace = "test.com.ar/user/v1"
        aliases = ["User", "My favorite User"]
```

--------------------------------

### Configure Pydantic model enum values

Source: https://marcosschroh.github.io/dataclasses-avroschema/pydantic

Demonstrates the difference between default enum serialization and using ConfigDict to force enum value serialization.

```python
import enum
from dataclasses_avroschema.pydantic import AvroBaseModel

class Color(str, enum.Enum):
    BLUE = "BLUE"
    RED = "RED"


class Bus(AvroBaseModel):
    driver: str
    color: Color

bus =  Bus(driver="bond", color=Color.RED)
print(bus.dict())
# >>> {'driver': 'bond', 'color': <Color.RED: 'RED'>}
```

```python
import enum
from dataclasses_avroschema.pydantic import AvroBaseModel
from pydantic import ConfigDict


class Color(str, enum.Enum):
    BLUE = "BLUE"
    RED = "RED"


class Bus(AvroBaseModel):
    model_config = ConfigDict(use_enum_values=True)
    driver: str
    color: Color

bus =  Bus(driver="bond", color=Color.RED)
print(busmodel_dump())
# >>> {'driver': 'bond', 'color': 'RED'}
```

--------------------------------

### Handling Malformed Avro Schema with Name Conflicts

Source: https://marcosschroh.github.io/dataclasses-avroschema/model_generator

When a field name in an Avro schema matches a record name, it can lead to annotation overshadowing in the generated Python models. This example shows the problematic output.

```json
{
  "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,
    }
  ]
}

```

```python
from dataclasses_avroschema import AvroModel
import dataclasses
import typing


@dataclasses.dataclass
class Address(AvroModel):
    name: str


@dataclasses.dataclass
class User(AvroModel):
    name: str
    age: int
    Address: typing.Optional[Address] = None

# Address` is `overshadowed` !!!
print(User.__annotations__)
# >>> {'name': str, 'age': int, 'Address': NoneType}

# We do not want this!!!
print(User.fake())
# >>> User(name='ftXgdDSUzdUIamiiHOiS', age=2422, Address=None)

```
Tokens:
Raw