feat: support for typing.Annotated added

README.md

@@ -329,8 +329,9 @@ User.fake()
 
 ## 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] 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] Logical Types: date, time (millis and micro), datetime (millis and micro), uuid support
 * [X] Schema relations (oneToOne, oneToMany)
 * [X] Recursive Schemas

dataclasses_avroschema/field_utils.py

@@ -91,7 +91,17 @@
 }
 
 # excluding tuple because is a container
-PYTHON_INMUTABLE_TYPES = (str, int, types.Int32, types.Float32, bool, float, bytes, type(None))
+PYTHON_INMUTABLE_TYPES = (
+    str,
+    int,
+    types.Int32,
+    types.Float32,
+    bool,
+    float,
+    bytes,
+    type(None),
+)
+
 PYTHON_PRIMITIVE_CONTAINERS = (list, tuple, dict)
 
 PYTHON_LOGICAL_TYPES = (

dataclasses_avroschema/fields.py

@@ -16,6 +16,7 @@
 import inflect
 from faker import Faker
 from pytz import utc
+from typing_extensions import get_args
 
 from dataclasses_avroschema import schema_generator, serialization, types, utils
 
@@ -49,6 +50,7 @@ class BaseField:
     type: typing.Any  # store the python primitive type
     default: typing.Any
     parent: typing.Any
+    field_info: typing.Optional[types.FieldInfo] = None
     metadata: typing.Optional[typing.Mapping] = None
     model_metadata: typing.Optional[utils.SchemaMetadata] = None
 
@@ -122,11 +124,12 @@ def get_default_value(self) -> typing.Any:
             return self.default
 
     def validate_default(self) -> bool:
+        a_type = self.type
         msg = f"Invalid default type. Default should be {self.type}"
-        if getattr(self.type, "__metadata__", [None])[0] in types.CUSTOM_TYPES:
-            assert isinstance(self.default, self.type.__origin__)
-        else:
-            assert isinstance(self.default, self.type), msg
+        if utils.is_annotated(self.type):
+            a_type, _ = get_args(self.type)
+
+        assert isinstance(self.default, a_type), msg
 
         return True
 
@@ -929,12 +932,21 @@ def field_factory(
     *,
     default: typing.Any = dataclasses.MISSING,
     default_factory: typing.Any = dataclasses.MISSING,
-    metadata: typing.Optional[typing.Mapping] = None,
+    metadata: typing.Optional[typing.Dict[str, typing.Any]] = None,
     model_metadata: typing.Optional[utils.SchemaMetadata] = None,
 ) -> FieldType:
     if metadata is None:
         metadata = {}
 
+    field_info = None
+
+    if utils.is_annotated(native_type):
+        a_type, *extra_args = get_args(native_type)
+        field_info = next((arg for arg in extra_args if isinstance(arg, types.FieldInfo)), None)
+
+        if field_info is None:
+            native_type = a_type
+
     if native_type in field_utils.PYTHON_INMUTABLE_TYPES:
         klass = INMUTABLE_FIELDS_CLASSES[native_type]
         return klass(
@@ -944,6 +956,7 @@ def field_factory(
             metadata=metadata,
             model_metadata=model_metadata,
             parent=parent,
+            field_info=field_info,
         )
     elif utils.is_self_referenced(native_type):
         return SelfReferenceField(
@@ -953,6 +966,7 @@ def field_factory(
             metadata=metadata,
             model_metadata=model_metadata,
             parent=parent,
+            field_info=field_info,
         )
     elif native_type is types.Fixed:
         return FixedField(
@@ -962,6 +976,7 @@ def field_factory(
             metadata=metadata,
             model_metadata=model_metadata,
             parent=parent,
+            field_info=field_info,
         )
     elif native_type in field_utils.PYTHON_LOGICAL_TYPES:
         klass = LOGICAL_TYPES_FIELDS_CLASSES[native_type]  # type: ignore
@@ -973,6 +988,7 @@ def field_factory(
             metadata=metadata,
             model_metadata=model_metadata,
             parent=parent,
+            field_info=field_info,
         )
     elif isinstance(native_type, GenericAlias):  # type: ignore
         origin = native_type.__origin__
@@ -1003,6 +1019,7 @@ def field_factory(
             default_factory=default_factory,
             model_metadata=model_metadata,
             parent=parent,
+            field_info=field_info,
         )
     elif inspect.isclass(native_type) and issubclass(native_type, enum.Enum):
         return EnumField(
@@ -1012,6 +1029,7 @@ def field_factory(
             metadata=metadata,
             model_metadata=model_metadata,
             parent=parent,
+            field_info=field_info,
         )
     elif types.UnionType is not None and isinstance(native_type, types.UnionType):
         # we need to check whether types.UnionType because it works only in
@@ -1026,6 +1044,7 @@ def field_factory(
             default_factory=default_factory,
             model_metadata=model_metadata,
             parent=parent,
+            field_info=field_info,
         )
     elif inspect.isclass(native_type) and issubclass(native_type, schema_generator.AvroModel):
         return RecordField(
@@ -1035,6 +1054,7 @@ def field_factory(
             metadata=metadata,
             model_metadata=model_metadata,
             parent=parent,
+            field_info=field_info,
         )
     else:
         msg = (

dataclasses_avroschema/schema_definition.py

@@ -68,7 +68,7 @@ def parse_fields(self) -> typing.List[FieldType]:
                 dataclass_field.type,
                 default=dataclass_field.default,
                 default_factory=dataclass_field.default_factory,  # type: ignore  # TODO: resolve mypy
-                metadata=dataclass_field.metadata,
+                metadata=dict(dataclass_field.metadata),
                 model_metadata=self.metadata,
                 parent=self.parent,
             )
@@ -100,7 +100,7 @@ def parse_faust_fields(self) -> typing.List[FieldType]:
                         dataclass_field.type,
                         default=default,
                         default_factory=default_factory,
-                        metadata=metadata,
+                        metadata=dict(metadata),
                         model_metadata=self.metadata,
                         parent=self.parent,
                     )

dataclasses_avroschema/types.py

@@ -28,6 +28,21 @@
 __all__ = CUSTOM_TYPES
 
 
+class FieldInfo:
+    def __init__(self, **kwargs) -> None:
+        self.type = kwargs.get("type")
+        self.max_digits = kwargs.get("max_digits")
+        self.decimal_places = kwargs.get("decimal_places")
+
+    @property
+    def metadata(self):
+        return {
+            "type": self.type,
+            "max_digits": self.max_digits,
+            "decimal_places": self.decimal_places,
+        }
+
+
 class MissingSentinel(typing.Generic[T]):
     """
     Class to detect when a field is not initialized
@@ -80,7 +95,7 @@ def __repr__(self) -> str:
         return f"Decimal('{self.default}')"
 
 
-Int32 = Annotated[int, "Int32"]
-Float32 = Annotated[float, "Float32"]
-TimeMicro = Annotated[datetime.time, "TimeMicro"]
-DateTimeMicro = Annotated[datetime.datetime, "DateTimeMicro"]
+Int32 = Annotated[int, FieldInfo(type="Int32")]
+Float32 = Annotated[float, FieldInfo(type="Float32")]
+TimeMicro = Annotated[datetime.time, FieldInfo(type="TimeMicro")]
+DateTimeMicro = Annotated[datetime.datetime, FieldInfo(type="DateTimeMicro")]

dataclasses_avroschema/utils.py

@@ -3,6 +3,7 @@
 from datetime import datetime
 
 from pytz import utc
+from typing_extensions import Annotated, get_origin
 
 from .types import JsonDict
 
@@ -69,6 +70,11 @@ def is_self_referenced(a_type: type) -> bool:
     )
 
 
+def is_annotated(a_type: typing.Any) -> bool:
+    origin = get_origin(a_type)
+    return origin is not None and isinstance(origin, type) and issubclass(origin, Annotated)  # type: ignore[arg-type]
+
+
 @dataclasses.dataclass
 class SchemaMetadata:
     schema_name: typing.Optional[str] = None

docs/fields_specification.md

@@ -137,6 +137,62 @@ Python Type | Avro Type   | Logical Type |
 | uuid.uuid4  | string    |  uuid        |
 | uuid.UUID    |  string        | uuid |
 
+## typing.Annotated
+
+All the types can be [Annotated](https://docs.python.org/3/library/typing.html#typing.Annotated) so `metadata` can be added to the fields. This library will use the `python type` to generate the `avro field` and it will ignore the extra `metadata`.
+
+```python title="Annotated"
+import dataclasses
+import enum
+import typing
+
+from dataclasses_avroschema import AvroModel
+
+
+class FavoriteColor(str, enum.Enum):
+    BLUE = "BLUE"
+    YELLOW = "YELLOW"
+    GREEN = "GREEN"
+
+
+@dataclasses.dataclass
+class UserAdvance(AvroModel):
+    name: typing.Annotated[str, "string"]
+    age: typing.Annotated[int, "integer"]
+    pets: typing.List[typing.Annotated[str, "string"]]
+    accounts: typing.Dict[str, typing.Annotated[int, "integer"]]
+    favorite_colors: typing.Annotated[FavoriteColor, "a color enum"]
+    has_car: typing.Annotated[bool, "boolean"] = False
+    country: str = "Argentina"
+    address: typing.Optional[typing.Annotated[str, "string"]] = None
+
+    class Meta:
+        schema_doc = False
+
+
+UserAdvance.avro_schema()
+```
+
+resulting in
+
+```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": "favorite_colors", "type": {"type": "enum", "name": "FavoriteColor", "symbols": ["BLUE", "YELLOW", "GREEN"]}},
+    {"name": "has_car", "type": "boolean", "default": false}, 
+    {"name": "country", "type": "string", "default": "Argentina"},
+    {"name": "address", "type": ["null", "string"], "default": null}]
+}'
+```
+
+*(This script is complete, it should run "as is")*
+
 ## Adding Custom Field-level Attributes
 
 You may want to add field-level attributes which are not automatically populated according to the typing semantics
@@ -149,8 +205,6 @@ to all fields such as `"name"` and others are specific to the datatype (e.g. `ar
 In order to add custom fields, you can use the `field` descriptor of the built-in `dataclasses` package and provide a
 `dict` of key-value pairs to the `metadata` parameter as in `dataclasses.field(metadata={'doc': 'foo'})`.
 
-### Examples
-
 ```python title="Adding a doc attribute to fields"
 from dataclasses import dataclass, field
 from dataclasses_avroschema import AvroModel, types

tests/conftest.py

@@ -4,6 +4,7 @@
 import typing
 
 import pytest
+from typing_extensions import Annotated
 
 from dataclasses_avroschema import AvroModel, types
 
@@ -154,7 +155,7 @@ class UserAdvance(AvroModel):
         favorite_colors: color_enum
         has_car: bool = False
         country: str = "Argentina"
-        address: str = None
+        address: typing.Optional[str] = None
         user_type: typing.Union[int, user_type_enum] = -1
         md5: types.Fixed = types.Fixed(16)
 
@@ -164,6 +165,27 @@ class Meta:
     return UserAdvance
 
 
+@pytest.fixture
+def user_advance_dataclass_with_union_enum_with_annotated(color_enum: type, user_type_enum: type):
+    @dataclasses.dataclass
+    class UserAdvance(AvroModel):
+        name: Annotated[str, "string"]
+        age: Annotated[int, "integer"]
+        pets: typing.List[Annotated[str, "string"]]
+        accounts: typing.Dict[str, Annotated[int, "integer"]]
+        favorite_colors: Annotated[color_enum, "a color enum"]
+        has_car: Annotated[bool, "boolean"] = False
+        country: str = "Argentina"
+        address: typing.Optional[Annotated[str, "string"]] = None
+        user_type: typing.Union[Annotated[int, "integer"], user_type_enum] = -1
+        md5: types.Fixed = types.Fixed(16)
+
+        class Meta:
+            schema_doc = False
+
+    return UserAdvance
+
+
 @pytest.fixture
 def user_advance_dataclass_with_sub_record_and_enum(color_enum: type, user_type_enum: type):
     @dataclasses.dataclass