Skip to content

API Reference

dynapydantic - dynamic tracking of pydantic models

AmbiguousDiscriminatorValueError

Bases: Error

Occurs when the discriminator value is ambiguous

Source code in src/dynapydantic/exceptions.py 
12
13
class AmbiguousDiscriminatorValueError(Error):
    """Occurs when the discriminator value is ambiguous"""

ConfigurationError

Bases: Error

Occurs when the user misconfigured a tracking setup

Source code in src/dynapydantic/exceptions.py 
16
17
class ConfigurationError(Error):
    """Occurs when the user misconfigured a tracking setup"""

Error

Bases: Exception

Base class for all dynapydanitc errors

Source code in src/dynapydantic/exceptions.py 
4
5
class Error(Exception):
    """Base class for all dynapydanitc errors"""

RegistrationError

Bases: Error

Occurs when a model cannot be registered

Source code in src/dynapydantic/exceptions.py 
8
9
class RegistrationError(Error):
    """Occurs when a model cannot be registered"""

SubclassTrackingModel pydantic-model

Bases: BaseModel

Subclass-tracking BaseModel

This will inject a TrackingGroup into your class and automate the registration of subclasses.

Inheriting from this class will augment your class with the following members functions: 1. registered_subclasses() -> dict[str, type[cls]]: This will return a mapping of discriminator value to the corresponding sublcass. See TrackingGroup.models for details. 2. union() -> typing.GenericAlias: This will return an (optionally) annotated subclass union. See TrackingGroup.union() for details. 3. load_plugins() -> None: If plugin_entry_point was specified, then this method will load plugin packages to discover additional subclasses. See TrackingGroup.load_plugins for more details.

Source code in src/dynapydantic/subclass_tracking_model.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
class SubclassTrackingModel(pydantic.BaseModel):
    """Subclass-tracking BaseModel

    This will inject a TrackingGroup into your class and automate the
    registration of subclasses.

    Inheriting from this class will augment your class with the following
    members functions:
    1. registered_subclasses() -> dict[str, type[cls]]:
        This will return a mapping of discriminator value to the corresponding
        sublcass. See TrackingGroup.models for details.
    2. union() -> typing.GenericAlias:
        This will return an (optionally) annotated subclass union. See
        TrackingGroup.union() for details.
    3. load_plugins() -> None:
        If plugin_entry_point was specified, then this method will load plugin
        packages to discover additional subclasses. See
        TrackingGroup.load_plugins for more details.
    """

    def __init_subclass__(
        cls,
        *args,
        exclude_from_union: bool | None = None,
        **kwargs,
    ) -> None:
        """Subclass hook"""
        # Intercept any kwargs that are intended for TrackingGroup
        super().__pydantic_init_subclass__(
            *args,
            **{k: v for k, v in kwargs.items() if k not in TrackingGroup.model_fields},
        )

    @classmethod
    def __pydantic_init_subclass__(
        cls,
        *args,
        exclude_from_union: bool | None = None,
        **kwargs,
    ) -> None:
        """Pydantic subclass hook"""
        if SubclassTrackingModel in cls.__bases__:
            # Intercept any kwargs that are intended for TrackingGroup
            super().__pydantic_init_subclass__(
                *args,
                **{
                    k: v
                    for k, v in kwargs.items()
                    if k not in TrackingGroup.model_fields
                },
            )

            if isinstance(getattr(cls, "tracking_config", None), TrackingGroup):
                cls.__DYNAPYDANTIC__ = cls.tracking_config
            else:
                try:
                    cls.__DYNAPYDANTIC__: TrackingGroup = TrackingGroup.model_validate(
                        {"name": f"{cls.__name__}-subclasses"} | kwargs,
                    )
                except pydantic.ValidationError as e:
                    msg = (
                        "SubclassTrackingModel subclasses must either have a "
                        "tracking_config: ClassVar[dynapydantic.TrackingGroup] "
                        "member or pass kwargs sufficient to construct a "
                        "dynapydantic.TrackingGroup in the class declaration. "
                        "The latter approach produced the following "
                        f"ValidationError:\n{e}"
                    )
                    raise ConfigurationError(msg) from e

            # Promote the tracking group's methods to the parent class
            if cls.__DYNAPYDANTIC__.plugin_entry_point is not None:

                def _load_plugins() -> None:
                    """Load plugins to register more models"""
                    cls.__DYNAPYDANTIC__.load_plugins()

                cls.load_plugins = staticmethod(_load_plugins)

            def _union(*, annotated: bool = True) -> ty.GenericAlias:
                """Get the union of all tracked subclasses

                Parameters
                ----------
                annotated
                    Whether this should be an annotated union for usage as a
                    pydantic field annotation, or a plain typing.Union for a
                    regular type annotation.
                """
                return cls.__DYNAPYDANTIC__.union(annotated=annotated)

            cls.union = staticmethod(_union)

            def _subclasses() -> dict[str, type[cls]]:
                """Return a mapping of discriminator values to registered model"""
                return cls.__DYNAPYDANTIC__.models

            cls.registered_subclasses = staticmethod(_subclasses)

            return

        super().__pydantic_init_subclass__(*args, **kwargs)

        if exclude_from_union:
            return

        supers = direct_children_of_base_in_mro(cls, SubclassTrackingModel)
        for base in supers:
            base.__DYNAPYDANTIC__.register_model(cls)

__init_subclass__(*args, exclude_from_union=None, **kwargs)

Subclass hook

Source code in src/dynapydantic/subclass_tracking_model.py 
48
49
50
51
52
53
54
55
56
57
58
59
def __init_subclass__(
    cls,
    *args,
    exclude_from_union: bool | None = None,
    **kwargs,
) -> None:
    """Subclass hook"""
    # Intercept any kwargs that are intended for TrackingGroup
    super().__pydantic_init_subclass__(
        *args,
        **{k: v for k, v in kwargs.items() if k not in TrackingGroup.model_fields},
    )

__pydantic_init_subclass__(*args, exclude_from_union=None, **kwargs) classmethod

Pydantic subclass hook

Source code in src/dynapydantic/subclass_tracking_model.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
@classmethod
def __pydantic_init_subclass__(
    cls,
    *args,
    exclude_from_union: bool | None = None,
    **kwargs,
) -> None:
    """Pydantic subclass hook"""
    if SubclassTrackingModel in cls.__bases__:
        # Intercept any kwargs that are intended for TrackingGroup
        super().__pydantic_init_subclass__(
            *args,
            **{
                k: v
                for k, v in kwargs.items()
                if k not in TrackingGroup.model_fields
            },
        )

        if isinstance(getattr(cls, "tracking_config", None), TrackingGroup):
            cls.__DYNAPYDANTIC__ = cls.tracking_config
        else:
            try:
                cls.__DYNAPYDANTIC__: TrackingGroup = TrackingGroup.model_validate(
                    {"name": f"{cls.__name__}-subclasses"} | kwargs,
                )
            except pydantic.ValidationError as e:
                msg = (
                    "SubclassTrackingModel subclasses must either have a "
                    "tracking_config: ClassVar[dynapydantic.TrackingGroup] "
                    "member or pass kwargs sufficient to construct a "
                    "dynapydantic.TrackingGroup in the class declaration. "
                    "The latter approach produced the following "
                    f"ValidationError:\n{e}"
                )
                raise ConfigurationError(msg) from e

        # Promote the tracking group's methods to the parent class
        if cls.__DYNAPYDANTIC__.plugin_entry_point is not None:

            def _load_plugins() -> None:
                """Load plugins to register more models"""
                cls.__DYNAPYDANTIC__.load_plugins()

            cls.load_plugins = staticmethod(_load_plugins)

        def _union(*, annotated: bool = True) -> ty.GenericAlias:
            """Get the union of all tracked subclasses

            Parameters
            ----------
            annotated
                Whether this should be an annotated union for usage as a
                pydantic field annotation, or a plain typing.Union for a
                regular type annotation.
            """
            return cls.__DYNAPYDANTIC__.union(annotated=annotated)

        cls.union = staticmethod(_union)

        def _subclasses() -> dict[str, type[cls]]:
            """Return a mapping of discriminator values to registered model"""
            return cls.__DYNAPYDANTIC__.models

        cls.registered_subclasses = staticmethod(_subclasses)

        return

    super().__pydantic_init_subclass__(*args, **kwargs)

    if exclude_from_union:
        return

    supers = direct_children_of_base_in_mro(cls, SubclassTrackingModel)
    for base in supers:
        base.__DYNAPYDANTIC__.register_model(cls)

TrackingGroup pydantic-model

Bases: BaseModel

Tracker for pydantic models

Fields:

Source code in src/dynapydantic/tracking_group.py 
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
class TrackingGroup(pydantic.BaseModel):
    """Tracker for pydantic models"""

    name: str = pydantic.Field(
        description=(
            "Name of the tracking group. This is for human display, so it "
            "doesn't technically need to be globally unique, but it should be "
            "meaningfully named, as it will be used in error messages."
        ),
    )
    discriminator_field: str = pydantic.Field(
        description="Name of the discriminator field",
    )
    plugin_entry_point: str | None = pydantic.Field(
        None,
        description=(
            "If given, then plugins packages will be supported through this "
            "Python entrypoint. The entrypoint can either be a function, "
            "which will be called, or simply a module, which will be "
            "imported. In either case, models found along the import path of "
            "the entrypoint will be registered. If the entrypoint is a "
            "function, additional models may be declared in the function."
        ),
    )
    discriminator_value_generator: ty.Callable[[type], str] | None = pydantic.Field(
        None,
        description=(
            "A callable that produces default values for the discriminator field"
        ),
    )
    models: dict[str, type[pydantic.BaseModel]] = pydantic.Field(
        {},
        description="The tracked models",
    )

    def load_plugins(self) -> None:
        """Load plugins to discover/register additional models"""
        if self.plugin_entry_point is None:
            return

        from importlib.metadata import entry_points  # noqa: PLC0415

        for ep in entry_points().select(group=self.plugin_entry_point):
            plugin = ep.load()
            if callable(plugin):
                plugin()

    def register(
        self,
        discriminator_value: str | None = None,
    ) -> ty.Callable[[type], type]:
        """Register a model into this group (decorator)

        Parameters
        ----------
        discriminator_value
            Value for the discriminator field. If not given, then
            discriminator_value_generator must be non-None or the
            discriminator field must be declared by hand.
        """

        def _wrapper(cls: type[pydantic.BaseModel]) -> None:
            disc = self.discriminator_field
            field = cls.model_fields.get(self.discriminator_field)
            if field is None:
                if discriminator_value is not None:
                    _inject_discriminator_field(cls, disc, discriminator_value)
                elif self.discriminator_value_generator is not None:
                    _inject_discriminator_field(
                        cls,
                        disc,
                        self.discriminator_value_generator(cls),
                    )
                else:
                    msg = (
                        f"unable to determine a discriminator value for "
                        f'{cls.__name__} in tracking group "{self.name}". No '
                        "value was passed to register(), "
                        "discriminator_value_generator was None and the "
                        f'"{disc}" field was not defined.'
                    )
                    raise RegistrationError(msg)
            elif (
                discriminator_value is not None and field.default != discriminator_value
            ):
                msg = (
                    f"the discriminator value for {cls.__name__} was "
                    f'ambiguous, it was set to "{discriminator_value}" via '
                    f'register() and "{field.default}" via the discriminator '
                    f"field ({self.discriminator_field})."
                )
                raise AmbiguousDiscriminatorValueError(msg)

            self._register_with_discriminator_field(cls)
            return cls

        return _wrapper

    def register_model(self, cls: type[pydantic.BaseModel]) -> None:
        """Register the given model into this group

        Parameters
        ----------
        cls
            The model to register
        """
        disc = self.discriminator_field
        if cls.model_fields.get(self.discriminator_field) is None:
            if self.discriminator_value_generator is not None:
                _inject_discriminator_field(
                    cls,
                    disc,
                    self.discriminator_value_generator(cls),
                )
            else:
                msg = (
                    f"unable to determine a discriminator value for "
                    f'{cls.__name__} in tracking group "{self.name}", '
                    "discriminator_value_generator was None and the "
                    f'"{disc}" field was not defined.'
                )
                raise RegistrationError(msg)

        self._register_with_discriminator_field(cls)

    def _register_with_discriminator_field(self, cls: type[pydantic.BaseModel]) -> None:
        """Register the model with the default of the discriminator field

        Parameters
        ----------
        cls
            The class to register, must have the disciminator field set with a
            unique default value in the group.
        """
        disc = self.discriminator_field
        field = cls.model_fields.get(disc)
        value = field.default
        if value == pydantic_core.PydanticUndefined:
            msg = (
                f"{cls.__name__}.{disc} had no default value, it must "
                "have one which is unique among all tracked models."
            )
            raise RegistrationError(msg)

        if (other := self.models.get(value)) is not None and other is not cls:
            msg = (
                f'Cannot register {cls.__name__} under the "{value}" '
                f"identifier, which is already in use by {other.__name__}."
            )
            raise RegistrationError(msg)

        self.models[value] = cls

    def union(self, *, annotated: bool = True) -> ty.GenericAlias:
        """Return the union of all registered models"""
        return (
            ty.Annotated[
                ty.Union[  # noqa: UP007
                    tuple(
                        ty.Annotated[x, pydantic.Tag(v)] for v, x in self.models.items()
                    )
                ],
                pydantic.Field(discriminator=self.discriminator_field),
            ]
            if annotated
            else ty.Union[tuple(self.models.values())]  # noqa: UP007
        )

discriminator_field pydantic-field

Name of the discriminator field

discriminator_value_generator = None pydantic-field

A callable that produces default values for the discriminator field

models = {} pydantic-field

The tracked models

name pydantic-field

Name of the tracking group. This is for human display, so it doesn't technically need to be globally unique, but it should be meaningfully named, as it will be used in error messages.

plugin_entry_point = None pydantic-field

If given, then plugins packages will be supported through this Python entrypoint. The entrypoint can either be a function, which will be called, or simply a module, which will be imported. In either case, models found along the import path of the entrypoint will be registered. If the entrypoint is a function, additional models may be declared in the function.

load_plugins()

Load plugins to discover/register additional models

Source code in src/dynapydantic/tracking_group.py 
72
73
74
75
76
77
78
79
80
81
82
def load_plugins(self) -> None:
    """Load plugins to discover/register additional models"""
    if self.plugin_entry_point is None:
        return

    from importlib.metadata import entry_points  # noqa: PLC0415

    for ep in entry_points().select(group=self.plugin_entry_point):
        plugin = ep.load()
        if callable(plugin):
            plugin()

register(discriminator_value=None)

Register a model into this group (decorator)

Parameters:

Name Type Description Default
discriminator_value str | None

Value for the discriminator field. If not given, then discriminator_value_generator must be non-None or the discriminator field must be declared by hand.

 None
Source code in src/dynapydantic/tracking_group.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def register(
    self,
    discriminator_value: str | None = None,
) -> ty.Callable[[type], type]:
    """Register a model into this group (decorator)

    Parameters
    ----------
    discriminator_value
        Value for the discriminator field. If not given, then
        discriminator_value_generator must be non-None or the
        discriminator field must be declared by hand.
    """

    def _wrapper(cls: type[pydantic.BaseModel]) -> None:
        disc = self.discriminator_field
        field = cls.model_fields.get(self.discriminator_field)
        if field is None:
            if discriminator_value is not None:
                _inject_discriminator_field(cls, disc, discriminator_value)
            elif self.discriminator_value_generator is not None:
                _inject_discriminator_field(
                    cls,
                    disc,
                    self.discriminator_value_generator(cls),
                )
            else:
                msg = (
                    f"unable to determine a discriminator value for "
                    f'{cls.__name__} in tracking group "{self.name}". No '
                    "value was passed to register(), "
                    "discriminator_value_generator was None and the "
                    f'"{disc}" field was not defined.'
                )
                raise RegistrationError(msg)
        elif (
            discriminator_value is not None and field.default != discriminator_value
        ):
            msg = (
                f"the discriminator value for {cls.__name__} was "
                f'ambiguous, it was set to "{discriminator_value}" via '
                f'register() and "{field.default}" via the discriminator '
                f"field ({self.discriminator_field})."
            )
            raise AmbiguousDiscriminatorValueError(msg)

        self._register_with_discriminator_field(cls)
        return cls

    return _wrapper

register_model(cls)

Register the given model into this group

Parameters:

Name Type Description Default
cls type[BaseModel]

The model to register

 required
Source code in src/dynapydantic/tracking_group.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def register_model(self, cls: type[pydantic.BaseModel]) -> None:
    """Register the given model into this group

    Parameters
    ----------
    cls
        The model to register
    """
    disc = self.discriminator_field
    if cls.model_fields.get(self.discriminator_field) is None:
        if self.discriminator_value_generator is not None:
            _inject_discriminator_field(
                cls,
                disc,
                self.discriminator_value_generator(cls),
            )
        else:
            msg = (
                f"unable to determine a discriminator value for "
                f'{cls.__name__} in tracking group "{self.name}", '
                "discriminator_value_generator was None and the "
                f'"{disc}" field was not defined.'
            )
            raise RegistrationError(msg)

    self._register_with_discriminator_field(cls)

union(*, annotated=True)

Return the union of all registered models

Source code in src/dynapydantic/tracking_group.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
def union(self, *, annotated: bool = True) -> ty.GenericAlias:
    """Return the union of all registered models"""
    return (
        ty.Annotated[
            ty.Union[  # noqa: UP007
                tuple(
                    ty.Annotated[x, pydantic.Tag(v)] for v, x in self.models.items()
                )
            ],
            pydantic.Field(discriminator=self.discriminator_field),
        ]
        if annotated
        else ty.Union[tuple(self.models.values())]  # noqa: UP007
    )