| Name | dataclasses-avroschema JSON |
| Version |
0.65.5
JSON |
| download |
| home_page | None |
| Summary | Generate Avro Schemas from Python classes. Serialize/Deserialize python instances with avro schemas |
| upload_time | 2024-12-02 20:24:22 |
| maintainer | None |
| docs_url | None |
| author | Marcos Schroh |
| requires_python | <4.0,>=3.9 |
| license | MIT |
| keywords |
|
| VCS |
|
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# Dataclasses Avro Schema
Generate [avro schemas](https://avro.apache.org/docs/1.8.2/spec.html) from python dataclasses, [Pydantic](https://docs.pydantic.dev/latest/) models and [Faust](https://faust-streaming.github.io/faust/) Records. [Code generation](https://marcosschroh.github.io/dataclasses-avroschema/model_generator/) from avro schemas. [Serialize/Deserialize](https://marcosschroh.github.io/dataclasses-avroschema/serialization/) python instances with avro schemas
[](https://github.com/marcosschroh/dataclasses-avroschema/actions/workflows/tests.yaml)
[](https://github.com/marcosschroh/dataclasses-avroschema/blob/master/LICENSE)
[](https://codecov.io/gh/marcosschroh/dataclasses-avroschema)
## Requirements
`python 3.9+`
## Installation
with `pip` or `poetry`:
`pip install dataclasses-avroschema` or `poetry add dataclasses-avroschema`
### Extras
- [pydantic](https://docs.pydantic.dev/): `pip install 'dataclasses-avroschema[pydantic]'` or `poetry add dataclasses-avroschema --extras "pydantic"`
- [faust-streaming](https://github.com/faust-streaming/faust): `pip install 'dataclasses-avroschema[faust]'` or `poetry add dataclasses-avroschema --extras "faust"`
- [faker](https://github.com/joke2k/faker): `pip install 'dataclasses-avroschema[faker]'` or `poetry add dataclasses-avroschema --extras "faker"`
- [dc-avro](https://marcosschroh.github.io/dc-avro/): `pip install 'dataclasses-avroschema[cli]'` or `poetry add dataclasses-avroschema --with cli`
*Note*: You can install all extra dependencies with `pip install dataclasses-avroschema[faust,pydantic,faker,cli]` or `poetry add dataclasses-avroschema --extras "pydantic faust faker cli"`
## Documentation
https://marcosschroh.github.io/dataclasses-avroschema/
## Usage
### Generating the avro schema
```python
from dataclasses import dataclass
import enum
import typing
from dataclasses_avroschema import AvroModel
class FavoriteColor(str, enum.Enum):
BLUE = "BLUE"
YELLOW = "YELLOW"
GREEN = "GREEN"
@dataclass
class User(AvroModel):
"An User"
name: str
age: int
pets: typing.List[str]
accounts: typing.Dict[str, int]
favorite_colors: FavoriteColor
country: str = "Argentina"
address: typing.Optional[str] = None
class Meta:
namespace = "User.v1"
aliases = ["user-v1", "super user"]
print(User.avro_schema())
# {
# "type": "record",
# "name": "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_colors", "type": {"type": "enum", "name": "FavoriteColor", "symbols": ["BLUE", "YELLOW", "GREEN"]}},
# {"name": "country", "type": "string", "default": "Argentina"},
# {"name": "address", "type": ["null", "string"], "default": null}
# ],
# "doc": "An User",
# "namespace": "User.v1",
# "aliases": ["user-v1", "super user"]
# }
assert User.avro_schema_to_python() == {
"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_colors", "type": {"type": "enum", "name": "FavoriteColor", "symbols": ["BLUE", "YELLOW", "GREEN"]}},
{"name": "country", "type": "string", "default": "Argentina"},
{"name": "address", "type": ["null", "string"], "default": None}
],
}
```
### Serialization to avro or avro-json and json payload
For serialization is neccesary to use python class/dataclasses instance
```python
from dataclasses import dataclass
import typing
from dataclasses_avroschema import AvroModel
@dataclass
class Address(AvroModel):
"An Address"
street: str
street_number: int
@dataclass
class User(AvroModel):
"User with multiple Address"
name: str
age: int
addresses: typing.List[Address]
address_data = {
"street": "test",
"street_number": 10,
}
# create an Address instance
address = Address(**address_data)
data_user = {
"name": "john",
"age": 20,
"addresses": [address],
}
# create an User instance
user = User(**data_user)
# serialization
assert user.serialize() == b"\x08john(\x02\x08test\x14\x00"
assert user.serialize(
serialization_type="avro-json"
) == b'{"name": "john", "age": 20, "addresses": [{"street": "test", "street_number": 10}]}'
# # Get the json from the instance
assert user.to_json() == '{"name": "john", "age": 20, "addresses": [{"street": "test", "street_number": 10}]}'
# # Get a python dict
assert user.to_dict() == {
"name": "john",
"age": 20,
"addresses": [
{"street": "test", "street_number": 10}
]
}
```
### Deserialization
Deserialization could take place with an instance dataclass or the dataclass itself. Can return the dict representation or a new class instance
```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)]
)
# return a python dict
assert User.deserialize(avro_binary, create_instance=False) == {
"name": "john",
"age": 20,
"addresses": [
{"street": "test", "street_number": 10}
]
}
# 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)]
)
# 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}]}
```
## Pydantic integration
To add `dataclasses-avroschema` functionality to `pydantic` you only need to replace `BaseModel` by `AvroBaseModel`:
```python
import typing
import enum
from dataclasses_avroschema.pydantic import AvroBaseModel
from pydantic import Field, ValidationError
class FavoriteColor(str, enum.Enum):
BLUE = "BLUE"
YELLOW = "YELLOW"
GREEN = "GREEN"
class UserAdvance(AvroBaseModel):
name: str
age: int
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
favorite_colors: FavoriteColor = FavoriteColor.BLUE
country: str = "Argentina"
address: typing.Optional[str] = None
class Meta:
schema_doc = False
assert UserAdvance.avro_schema_to_python() == {
"type": "record",
"name": "UserAdvance",
"fields": [
{"name": "name", "type": "string"},
{"name": "age", "type": "long"},
{"name": "pets", "type": {"type": "array", "items": "string", "name": "pet"}, "default": ["dog", "cat"]},
{"name": "accounts", "type": {"type": "map", "values": "long", "name": "account"}, "default": {"key": 1}},
{"name": "has_car", "type": "boolean", "default": False},{"name": "favorite_colors", "type": {"type": "enum", "name": "FavoriteColor", "symbols": ["BLUE", "YELLOW", "GREEN"]}, "default": "BLUE"},
{"name": "country", "type": "string", "default": "Argentina"}, {"name": "address", "type": ["null", "string"], "default": None}
]
}
print(UserAdvance.json_schema())
# {
# "$defs": {"FavoriteColor": {"enum": ["BLUE", "YELLOW", "GREEN"], "title": "FavoriteColor", "type": "string"}},
# "properties": {
# "name": {"title": "Name", "type": "string"},
# "age": {"title": "Age", "type": "integer"},
# "pets": {"items": {"type": "string"}, "title": "Pets", "type": "array"},
# "accounts": {"additionalProperties": {"type": "integer"}, "title": "Accounts", "type": "object"},
# "has_car": {"default": false, "title": "Has Car", "type": "boolean"},
# "favorite_colors": {"allOf": [{"$ref": "#/$defs/FavoriteColor"}], "default": "BLUE"},
# "country": {"default": "Argentina", "title": "Country", "type": "string"},
# "address": {"anyOf": [{"type": "string"}, {"type": "null"}], "default": null, "title": "Address"}
# },
# "required": ["name", "age"],
# "title": "UserAdvance",
# "type": "object"
# }"""
user = UserAdvance(name="bond", age=50)
# pydantic
assert user.dict() == {
'name': 'bond',
'age': 50,
'pets': ['dog', 'cat'],
'accounts': {'key': 1},
'has_car': False,
'favorite_colors': FavoriteColor.BLUE,
'country': 'Argentina',
'address': None
}
# pydantic
print(user.json())
assert user.json() == '{"name":"bond","age":50,"pets":["dog","cat"],"accounts":{"key":1},"has_car":false,"favorite_colors":"BLUE","country":"Argentina","address":null}'
# pydantic
try:
user = UserAdvance(name="bond")
except ValidationError as exc:
...
# dataclasses-avroschema
event = user.serialize()
assert event == b'\x08bondd\x04\x06dog\x06cat\x00\x02\x06key\x02\x00\x00\x00\x12Argentina\x00'
assert UserAdvance.deserialize(data=event) == UserAdvance(
name='bond',
age=50,
pets=['dog', 'cat'],
accounts={'key': 1},
has_car=False,
favorite_colors=FavoriteColor.BLUE,
country='Argentina',
address=None
)
```
## Examples with python streaming drivers (kafka and redis)
Under [examples](https://github.com/marcosschroh/dataclasses-avroschema/tree/master/examples) folder you can find 3 differents kafka examples, one with [aiokafka](https://github.com/aio-libs/aiokafka) (`async`) showing the simplest use case when a `AvroModel` instance is serialized and sent it thorught kafka, and the event is consumed.
The other two examples are `sync` using the [kafka-python](https://github.com/dpkp/kafka-python) driver, where the `avro-json` serialization and `schema evolution` (`FULL` compatibility) is shown.
Also, there are two `redis` examples using `redis streams` with [walrus](https://github.com/coleifer/walrus) and [redisgears-py](https://github.com/RedisGears/redisgears-py)
## Factory and fixtures
[Dataclasses Avro Schema](https://github.com/marcosschroh/dataclasses-avroschema) also includes a `factory` feature, so you can generate `fast` python instances and use them, for example, to test your data streaming pipelines. Instances can be generated using the `fake` method.
*Note*: This feature is not enabled by default and requires you have the `faker` extra installed. You may install it with `pip install 'dataclasses-avroschema[faker]'`
```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)])
```
## Features
- [x] Primitive types: int, long, double, float, boolean, string and null support
- [x] Complex types: enum, array, map, fixed, unions and records support
- [x] `typing.Annotated` supported
- [x] `typing.Literal` supported
- [x] Logical Types: date, time (millis and micro), datetime (millis and micro), uuid support
- [x] Schema relations (oneToOne, oneToMany)
- [x] Recursive Schemas
- [x] Generate Avro Schemas from `faust.Record`
- [x] Instance serialization correspondent to `avro schema` generated
- [x] Data deserialization. Return python dict or class instance
- [x] Generate json from python class instance
- [x] Case Schemas
- [x] Generate models from `avsc` files
- [x] Examples of integration with `kafka` drivers: [aiokafka](https://github.com/aio-libs/aiokafka), [kafka-python](https://github.com/dpkp/kafka-python)
- [x] Example of integration with `redis` drivers: [walrus](https://github.com/coleifer/walrus) and [redisgears-py](https://github.com/RedisGears/redisgears-py)
- [x] Factory instances
- [x] [Pydantic](https://pydantic-docs.helpmanual.io/) integration
## Development
[Poetry](https://python-poetry.org/docs/) is needed to install the dependencies and develope locally
1. Install dependencies: `poetry install --all-extras`
2. Code linting: `./scripts/format`
3. Run tests: `./scripts/test`
4. Tests documentation: `./scripts/test-documentation`
For commit messages we use [commitizen](https://commitizen-tools.github.io/commitizen/) in order to standardize a way of committing rules
Raw data
{
"_id": null,
"home_page": null,
"name": "dataclasses-avroschema",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.9",
"maintainer_email": null,
"keywords": null,
"author": "Marcos Schroh",
"author_email": "schrohm@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/36/1b/0c3fe7be5265d6ba583c03b8a1f828f872ad4a4e1565aba3eb3009205456/dataclasses_avroschema-0.65.5.tar.gz",
"platform": null,
"description": "# Dataclasses Avro Schema\n\nGenerate [avro schemas](https://avro.apache.org/docs/1.8.2/spec.html) from python dataclasses, [Pydantic](https://docs.pydantic.dev/latest/) models and [Faust](https://faust-streaming.github.io/faust/) Records. [Code generation](https://marcosschroh.github.io/dataclasses-avroschema/model_generator/) from avro schemas. [Serialize/Deserialize](https://marcosschroh.github.io/dataclasses-avroschema/serialization/) python instances with avro schemas\n\n[](https://github.com/marcosschroh/dataclasses-avroschema/actions/workflows/tests.yaml)\n[](https://github.com/marcosschroh/dataclasses-avroschema/blob/master/LICENSE)\n[](https://codecov.io/gh/marcosschroh/dataclasses-avroschema)\n\n\n## Requirements\n\n`python 3.9+`\n\n## Installation\n\nwith `pip` or `poetry`:\n\n`pip install dataclasses-avroschema` or `poetry add dataclasses-avroschema`\n\n### Extras\n\n- [pydantic](https://docs.pydantic.dev/): `pip install 'dataclasses-avroschema[pydantic]'` or `poetry add dataclasses-avroschema --extras \"pydantic\"`\n- [faust-streaming](https://github.com/faust-streaming/faust): `pip install 'dataclasses-avroschema[faust]'` or `poetry add dataclasses-avroschema --extras \"faust\"`\n- [faker](https://github.com/joke2k/faker): `pip install 'dataclasses-avroschema[faker]'` or `poetry add dataclasses-avroschema --extras \"faker\"`\n- [dc-avro](https://marcosschroh.github.io/dc-avro/): `pip install 'dataclasses-avroschema[cli]'` or `poetry add dataclasses-avroschema --with cli`\n\n*Note*: You can install all extra dependencies with `pip install dataclasses-avroschema[faust,pydantic,faker,cli]` or `poetry add dataclasses-avroschema --extras \"pydantic faust faker cli\"`\n\n## Documentation\n\nhttps://marcosschroh.github.io/dataclasses-avroschema/\n\n## Usage\n\n### Generating the avro schema\n\n```python\nfrom dataclasses import dataclass\nimport enum\n\nimport typing\n\nfrom dataclasses_avroschema import AvroModel\n\n\nclass FavoriteColor(str, enum.Enum):\n BLUE = \"BLUE\"\n YELLOW = \"YELLOW\"\n GREEN = \"GREEN\"\n\n\n@dataclass\nclass User(AvroModel):\n \"An User\"\n name: str\n age: int\n pets: typing.List[str]\n accounts: typing.Dict[str, int]\n favorite_colors: FavoriteColor\n country: str = \"Argentina\"\n address: typing.Optional[str] = None\n\n class Meta:\n namespace = \"User.v1\"\n aliases = [\"user-v1\", \"super user\"]\n\n\nprint(User.avro_schema())\n\n# {\n# \"type\": \"record\",\n# \"name\": \"User\",\n# \"fields\": [\n# {\"name\": \"name\", \"type\": \"string\"},\n# {\"name\": \"age\", \"type\": \"long\"},\n# {\"name\": \"pets\", \"type\": {\"type\": \"array\", \"items\": \"string\", \"name\": \"pet\"}},\n# {\"name\": \"accounts\", \"type\": {\"type\": \"map\", \"values\": \"long\", \"name\": \"account\"}},\n# {\"name\": \"favorite_colors\", \"type\": {\"type\": \"enum\", \"name\": \"FavoriteColor\", \"symbols\": [\"BLUE\", \"YELLOW\", \"GREEN\"]}},\n# {\"name\": \"country\", \"type\": \"string\", \"default\": \"Argentina\"},\n# {\"name\": \"address\", \"type\": [\"null\", \"string\"], \"default\": null}\n# ], \n# \"doc\": \"An User\",\n# \"namespace\": \"User.v1\", \n# \"aliases\": [\"user-v1\", \"super user\"]\n# }\n\nassert User.avro_schema_to_python() == {\n \"type\": \"record\",\n \"name\": \"User\",\n \"doc\": \"An User\",\n \"namespace\": \"User.v1\",\n \"aliases\": [\"user-v1\", \"super user\"],\n \"fields\": [\n {\"name\": \"name\", \"type\": \"string\"},\n {\"name\": \"age\", \"type\": \"long\"},\n {\"name\": \"pets\", \"type\": {\"type\": \"array\", \"items\": \"string\", \"name\": \"pet\"}},\n {\"name\": \"accounts\", \"type\": {\"type\": \"map\", \"values\": \"long\", \"name\": \"account\"}},\n {\"name\": \"favorite_colors\", \"type\": {\"type\": \"enum\", \"name\": \"FavoriteColor\", \"symbols\": [\"BLUE\", \"YELLOW\", \"GREEN\"]}},\n {\"name\": \"country\", \"type\": \"string\", \"default\": \"Argentina\"},\n {\"name\": \"address\", \"type\": [\"null\", \"string\"], \"default\": None}\n ],\n}\n```\n\n### Serialization to avro or avro-json and json payload\n\nFor serialization is neccesary to use python class/dataclasses instance\n\n```python\nfrom dataclasses import dataclass\n\nimport typing\n\nfrom dataclasses_avroschema import AvroModel\n\n\n@dataclass\nclass Address(AvroModel):\n \"An Address\"\n street: str\n street_number: int\n\n\n@dataclass\nclass User(AvroModel):\n \"User with multiple Address\"\n name: str\n age: int\n addresses: typing.List[Address]\n\naddress_data = {\n \"street\": \"test\",\n \"street_number\": 10,\n}\n\n# create an Address instance\naddress = Address(**address_data)\n\ndata_user = {\n \"name\": \"john\",\n \"age\": 20,\n \"addresses\": [address],\n}\n\n# create an User instance\nuser = User(**data_user)\n\n# serialization\nassert user.serialize() == b\"\\x08john(\\x02\\x08test\\x14\\x00\"\n\nassert user.serialize(\n serialization_type=\"avro-json\"\n) == b'{\"name\": \"john\", \"age\": 20, \"addresses\": [{\"street\": \"test\", \"street_number\": 10}]}'\n\n# # Get the json from the instance\nassert user.to_json() == '{\"name\": \"john\", \"age\": 20, \"addresses\": [{\"street\": \"test\", \"street_number\": 10}]}'\n\n# # Get a python dict\nassert user.to_dict() == {\n \"name\": \"john\", \n \"age\": 20, \n \"addresses\": [\n {\"street\": \"test\", \"street_number\": 10}\n ]\n}\n```\n\n### Deserialization\n\nDeserialization could take place with an instance dataclass or the dataclass itself. Can return the dict representation or a new class instance\n\n```python\nimport typing\nimport dataclasses\n\nfrom dataclasses_avroschema import AvroModel\n\n\n@dataclasses.dataclass\nclass Address(AvroModel):\n \"An Address\"\n street: str\n street_number: int\n\n@dataclasses.dataclass\nclass User(AvroModel):\n \"User with multiple Address\"\n name: str\n age: int\n addresses: typing.List[Address]\n\navro_binary = b\"\\x08john(\\x02\\x08test\\x14\\x00\"\navro_json_binary = b'{\"name\": \"john\", \"age\": 20, \"addresses\": [{\"street\": \"test\", \"street_number\": 10}]}'\n\n# return a new class instance!!\nassert User.deserialize(avro_binary) == User(\n name='john', \n age=20,\n addresses=[Address(street='test', street_number=10)]\n)\n\n# return a python dict\nassert User.deserialize(avro_binary, create_instance=False) == {\n \"name\": \"john\",\n \"age\": 20,\n \"addresses\": [\n {\"street\": \"test\", \"street_number\": 10}\n ]\n}\n\n# return a new class instance!!\nassert User.deserialize(avro_json_binary, serialization_type=\"avro-json\") == User(\n name='john',\n age=20,\n addresses=[Address(street='test', street_number=10)]\n)\n\n# return a python dict\nassert User.deserialize(\n avro_json_binary,\n serialization_type=\"avro-json\",\n create_instance=False\n) == {\"name\": \"john\", \"age\": 20, \"addresses\": [{\"street\": \"test\", \"street_number\": 10}]}\n```\n\n## Pydantic integration\n\nTo add `dataclasses-avroschema` functionality to `pydantic` you only need to replace `BaseModel` by `AvroBaseModel`:\n\n```python\nimport typing\nimport enum\n\nfrom dataclasses_avroschema.pydantic import AvroBaseModel\n\nfrom pydantic import Field, ValidationError\n\n\nclass FavoriteColor(str, enum.Enum):\n BLUE = \"BLUE\"\n YELLOW = \"YELLOW\"\n GREEN = \"GREEN\"\n\n\nclass UserAdvance(AvroBaseModel):\n name: str\n age: int\n pets: typing.List[str] = Field(default_factory=lambda: [\"dog\", \"cat\"])\n accounts: typing.Dict[str, int] = Field(default_factory=lambda: {\"key\": 1})\n has_car: bool = False\n favorite_colors: FavoriteColor = FavoriteColor.BLUE\n country: str = \"Argentina\"\n address: typing.Optional[str] = None\n\n class Meta:\n schema_doc = False\n\n\nassert UserAdvance.avro_schema_to_python() == {\n \"type\": \"record\",\n \"name\": \"UserAdvance\",\n \"fields\": [\n {\"name\": \"name\", \"type\": \"string\"},\n {\"name\": \"age\", \"type\": \"long\"},\n {\"name\": \"pets\", \"type\": {\"type\": \"array\", \"items\": \"string\", \"name\": \"pet\"}, \"default\": [\"dog\", \"cat\"]},\n {\"name\": \"accounts\", \"type\": {\"type\": \"map\", \"values\": \"long\", \"name\": \"account\"}, \"default\": {\"key\": 1}},\n {\"name\": \"has_car\", \"type\": \"boolean\", \"default\": False},{\"name\": \"favorite_colors\", \"type\": {\"type\": \"enum\", \"name\": \"FavoriteColor\", \"symbols\": [\"BLUE\", \"YELLOW\", \"GREEN\"]}, \"default\": \"BLUE\"},\n {\"name\": \"country\", \"type\": \"string\", \"default\": \"Argentina\"}, {\"name\": \"address\", \"type\": [\"null\", \"string\"], \"default\": None}\n ]\n}\n\nprint(UserAdvance.json_schema())\n\n# {\n# \"$defs\": {\"FavoriteColor\": {\"enum\": [\"BLUE\", \"YELLOW\", \"GREEN\"], \"title\": \"FavoriteColor\", \"type\": \"string\"}},\n# \"properties\": {\n# \"name\": {\"title\": \"Name\", \"type\": \"string\"},\n# \"age\": {\"title\": \"Age\", \"type\": \"integer\"},\n# \"pets\": {\"items\": {\"type\": \"string\"}, \"title\": \"Pets\", \"type\": \"array\"},\n# \"accounts\": {\"additionalProperties\": {\"type\": \"integer\"}, \"title\": \"Accounts\", \"type\": \"object\"},\n# \"has_car\": {\"default\": false, \"title\": \"Has Car\", \"type\": \"boolean\"},\n# \"favorite_colors\": {\"allOf\": [{\"$ref\": \"#/$defs/FavoriteColor\"}], \"default\": \"BLUE\"},\n# \"country\": {\"default\": \"Argentina\", \"title\": \"Country\", \"type\": \"string\"},\n# \"address\": {\"anyOf\": [{\"type\": \"string\"}, {\"type\": \"null\"}], \"default\": null, \"title\": \"Address\"}\n# }, \n# \"required\": [\"name\", \"age\"],\n# \"title\": \"UserAdvance\",\n# \"type\": \"object\"\n# }\"\"\"\n\nuser = UserAdvance(name=\"bond\", age=50)\n\n# pydantic\nassert user.dict() == {\n 'name': 'bond',\n 'age': 50,\n 'pets': ['dog', 'cat'],\n 'accounts': {'key': 1},\n 'has_car': False,\n 'favorite_colors': FavoriteColor.BLUE,\n 'country': 'Argentina',\n 'address': None\n}\n\n# pydantic\nprint(user.json())\n\nassert user.json() == '{\"name\":\"bond\",\"age\":50,\"pets\":[\"dog\",\"cat\"],\"accounts\":{\"key\":1},\"has_car\":false,\"favorite_colors\":\"BLUE\",\"country\":\"Argentina\",\"address\":null}'\n\n# pydantic\ntry:\n user = UserAdvance(name=\"bond\")\nexcept ValidationError as exc:\n ...\n\n# dataclasses-avroschema\nevent = user.serialize()\nassert event == b'\\x08bondd\\x04\\x06dog\\x06cat\\x00\\x02\\x06key\\x02\\x00\\x00\\x00\\x12Argentina\\x00'\n\nassert UserAdvance.deserialize(data=event) == UserAdvance(\n name='bond',\n age=50, \n pets=['dog', 'cat'],\n accounts={'key': 1},\n has_car=False, \n favorite_colors=FavoriteColor.BLUE,\n country='Argentina', \n address=None\n)\n```\n\n## Examples with python streaming drivers (kafka and redis)\n\nUnder [examples](https://github.com/marcosschroh/dataclasses-avroschema/tree/master/examples) folder you can find 3 differents kafka examples, one with [aiokafka](https://github.com/aio-libs/aiokafka) (`async`) showing the simplest use case when a `AvroModel` instance is serialized and sent it thorught kafka, and the event is consumed.\nThe other two examples are `sync` using the [kafka-python](https://github.com/dpkp/kafka-python) driver, where the `avro-json` serialization and `schema evolution` (`FULL` compatibility) is shown.\nAlso, there are two `redis` examples using `redis streams` with [walrus](https://github.com/coleifer/walrus) and [redisgears-py](https://github.com/RedisGears/redisgears-py)\n\n## Factory and fixtures\n\n[Dataclasses Avro Schema](https://github.com/marcosschroh/dataclasses-avroschema) also includes a `factory` feature, so you can generate `fast` python instances and use them, for example, to test your data streaming pipelines. Instances can be generated using the `fake` method.\n\n*Note*: This feature is not enabled by default and requires you have the `faker` extra installed. You may install it with `pip install 'dataclasses-avroschema[faker]'`\n\n```python\nimport typing\nimport dataclasses\n\nfrom dataclasses_avroschema import AvroModel\n\n\n@dataclasses.dataclass\nclass Address(AvroModel):\n \"An Address\"\n street: str\n street_number: int\n\n\n@dataclasses.dataclass\nclass User(AvroModel):\n \"User with multiple Address\"\n name: str\n age: int\n addresses: typing.List[Address]\n\n\nAddress.fake()\n# >>>> Address(street='PxZJILDRgbXyhWrrPWxQ', street_number=2067)\n\nUser.fake()\n# >>>> User(name='VGSBbOGfSGjkMDnefHIZ', age=8974, addresses=[Address(street='vNpPYgesiHUwwzGcmMiS', street_number=4790)])\n```\n\n## Features\n\n- [x] Primitive types: int, long, double, float, boolean, string and null support\n- [x] Complex types: enum, array, map, fixed, unions and records support\n- [x] `typing.Annotated` supported\n- [x] `typing.Literal` supported\n- [x] Logical Types: date, time (millis and micro), datetime (millis and micro), uuid support\n- [x] Schema relations (oneToOne, oneToMany)\n- [x] Recursive Schemas\n- [x] Generate Avro Schemas from `faust.Record`\n- [x] Instance serialization correspondent to `avro schema` generated\n- [x] Data deserialization. Return python dict or class instance\n- [x] Generate json from python class instance\n- [x] Case Schemas\n- [x] Generate models from `avsc` files\n- [x] Examples of integration with `kafka` drivers: [aiokafka](https://github.com/aio-libs/aiokafka), [kafka-python](https://github.com/dpkp/kafka-python)\n- [x] Example of integration with `redis` drivers: [walrus](https://github.com/coleifer/walrus) and [redisgears-py](https://github.com/RedisGears/redisgears-py)\n- [x] Factory instances\n- [x] [Pydantic](https://pydantic-docs.helpmanual.io/) integration\n\n## Development\n\n[Poetry](https://python-poetry.org/docs/) is needed to install the dependencies and develope locally\n\n1. Install dependencies: `poetry install --all-extras`\n2. Code linting: `./scripts/format`\n3. Run tests: `./scripts/test`\n4. Tests documentation: `./scripts/test-documentation`\n\nFor commit messages we use [commitizen](https://commitizen-tools.github.io/commitizen/) in order to standardize a way of committing rules\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "Generate Avro Schemas from Python classes. Serialize/Deserialize python instances with avro schemas",
"version": "0.65.5",
"project_urls": null,
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "a189288fc6a8ed638cc2ed7398027b59a94caf3d1a8bfbbb5a1872686af57a01",
"md5": "7d6bec43671b52581a7ab7239719a36a",
"sha256": "a8643760547c0eccbbcedf919532738f1bfd80ff81854ebe4f52094b35a4a06f"
},
"downloads": -1,
"filename": "dataclasses_avroschema-0.65.5-py3-none-any.whl",
"has_sig": false,
"md5_digest": "7d6bec43671b52581a7ab7239719a36a",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.9",
"size": 57561,
"upload_time": "2024-12-02T20:24:20",
"upload_time_iso_8601": "2024-12-02T20:24:20.021137Z",
"url": "https://files.pythonhosted.org/packages/a1/89/288fc6a8ed638cc2ed7398027b59a94caf3d1a8bfbbb5a1872686af57a01/dataclasses_avroschema-0.65.5-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "361b0c3fe7be5265d6ba583c03b8a1f828f872ad4a4e1565aba3eb3009205456",
"md5": "c91b81ca6b0e69dc9b57aa656faf2647",
"sha256": "01e96fd66f7058094f7a0b86c072bf8e9d996de3c33622143774ea00a4803638"
},
"downloads": -1,
"filename": "dataclasses_avroschema-0.65.5.tar.gz",
"has_sig": false,
"md5_digest": "c91b81ca6b0e69dc9b57aa656faf2647",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.9",
"size": 45774,
"upload_time": "2024-12-02T20:24:22",
"upload_time_iso_8601": "2024-12-02T20:24:22.101769Z",
"url": "https://files.pythonhosted.org/packages/36/1b/0c3fe7be5265d6ba583c03b8a1f828f872ad4a4e1565aba3eb3009205456/dataclasses_avroschema-0.65.5.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-12-02 20:24:22",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "dataclasses-avroschema"
}
