Skip to content

API Reference

dynapydantic

dynapydantic - dynamic tracking of pydantic models

Polymorphic

Annotation used to mark a type as having duck-typing behavior

This annotation is only valid for SubclassTrackingModel's.

Similar to SerializeAsAny, a field annotated with this shall serialize as according to its actual type, not the field annotation type. In addition, parsing will function as if the field annotation type were the union of all tracked subclasses.

If a UnionRealization (or the string value of one) is passed as the second argument, it will override the default value for the union realization that is stored in the class.

Source code in src/dynapydantic/annotations.py
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
class Polymorphic:
    """Annotation used to mark a type as having duck-typing behavior

    This annotation is only valid for SubclassTrackingModel's.

    Similar to SerializeAsAny, a field annotated with this shall serialize as
    according to its actual type, not the field annotation type. In addition,
    parsing will function as if the field annotation type were the union of
    all tracked subclasses.

    If a UnionRealization (or the string value of one) is passed as the
    second argument, it will override the default value for the union
    realization that is stored in the class.
    """

    def __class_getitem__(
        cls,
        item: type[ModelT] | tuple[type[ModelT], UnionRealization | str],
    ) -> ty.Annotated[type[ModelT], ...]:
        """Get the annotation for the pydantic field"""
        if isinstance(item, tuple):
            if len(item) > 2:  # noqa: PLR2004
                msg = (
                    "dynapydantic.Polymorphic takes 1 or 2 arguments "
                    f"({len(item)} given)"
                )
                raise TypeError(msg)

            return _polymorphic_cgi(*item)
        return _polymorphic_cgi(item)

__class_getitem__(item: type[ModelT] | tuple[type[ModelT], UnionRealization | str]) -> ty.Annotated[type[ModelT], ...]

Get the annotation for the pydantic field

Source code in src/dynapydantic/annotations.py
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def __class_getitem__(
    cls,
    item: type[ModelT] | tuple[type[ModelT], UnionRealization | str],
) -> ty.Annotated[type[ModelT], ...]:
    """Get the annotation for the pydantic field"""
    if isinstance(item, tuple):
        if len(item) > 2:  # noqa: PLR2004
            msg = (
                "dynapydantic.Polymorphic takes 1 or 2 arguments "
                f"({len(item)} given)"
            )
            raise TypeError(msg)

        return _polymorphic_cgi(*item)
    return _polymorphic_cgi(item)

Union

Annotation used to get the union out of a dynapydantic entity

This annotation is primarily used for using the union of all models in a TrackingGroup as a field annotation. It can be used with SubclassTrackingModel, but in general, Polymorphic is preferable.

Source code in src/dynapydantic/annotations.py
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
class Union:
    """Annotation used to get the union out of a dynapydantic entity

    This annotation is primarily used for using the union of all models in
    a `TrackingGroup` as a field annotation. It can be used with
    `SubclassTrackingModel`, but in general, `Polymorphic` is preferable.
    """

    @ty.overload
    def __class_getitem__(cls, item: type[ModelT]) -> type[ModelT]: ...

    @ty.overload
    def __class_getitem__(cls, item: TrackingGroup) -> ty.Any: ...  # noqa: ANN401

    def __class_getitem__(cls, item: TrackingGroup | type[ModelT]) -> object:
        """Return the union"""
        return union(item)

__class_getitem__(item: TrackingGroup | type[ModelT]) -> object

__class_getitem__(item: type[ModelT]) -> type[ModelT]
__class_getitem__(item: TrackingGroup) -> ty.Any

Return the union

Source code in src/dynapydantic/annotations.py
103
104
105
def __class_getitem__(cls, item: TrackingGroup | type[ModelT]) -> object:
    """Return the union"""
    return union(item)

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

NoRegisteredTypesError

Bases: Error

Occurs when a union is requested from a tracking group with no members

Source code in src/dynapydantic/exceptions.py
20
21
class NoRegisteredTypesError(Error):
    """Occurs when a union is requested from a tracking group with no members"""

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.

Similar to BaseModel, SubclassTrackingModel can take arguments in the class declaration. Arguments from BaseModel will be forwarded. Additionally, any fields from TrackingGroup will be forwarded to the internal TrackingGroup instance. The following additional arguments are supported:

  1. exclude_from_union: This flag is intended to be used with descendents of SubclassTrackingModel. If True, this subclass will be omitted from tracking. The default for this flag is True for direct descendents of SubclassTrackingModel and False otherwise.
  2. union_realization: When the union should be realized. See UnionRealization for more details on the various options. The default is to realize unions at model construction time.
Source code in src/dynapydantic/subclass_tracking_model.py
16
17
18
19
20
21
22
23
24
25
26
27
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
class SubclassTrackingModel(pydantic.BaseModel):
    """Subclass-tracking BaseModel

    This will inject a [`TrackingGroup`][dynapydantic.TrackingGroup] into your
    class and automate the registration of subclasses.

    Similar to `BaseModel`, `SubclassTrackingModel` can take arguments in the
    class declaration. Arguments from `BaseModel` will be forwarded.
    Additionally, any fields from `TrackingGroup` will be forwarded to the
    internal `TrackingGroup` instance. The following additional arguments are
    supported:

    1. `exclude_from_union`: This flag is intended to be used with descendents
           of `SubclassTrackingModel`. If `True`, this subclass will be omitted
           from tracking. The default for this flag is `True` for direct
           descendents of `SubclassTrackingModel` and `False` otherwise.
    2. `union_realization`: When the union should be realized. See
           [`UnionRealization`][dynapydantic.UnionRealization] for more details
           on the various options. The default is to realize unions at model
           construction time.
    """

    def __init_subclass__(cls, *args, **kwargs) -> None:
        """Subclass hook"""
        # Intercept any kwargs that are intended for TrackingGroup or
        # __pydantic_init_subclass__
        sig = inspect.signature(SubclassTrackingModel.__pydantic_init_subclass__)
        super().__init_subclass__(
            *args,
            **{
                k: v
                for k, v in kwargs.items()
                if k not in TrackingGroup.model_fields and k not in sig.parameters
            },
        )

    @classmethod
    def __pydantic_init_subclass__(
        cls,
        *args,
        exclude_from_union: bool | None = None,
        union_realization: str | UnionRealization | None = None,
        **kwargs,
    ) -> None:
        """Pydantic subclass hook"""
        # Forward along any unexpected arguments that were not intended
        # for TrackingGroup.
        super().__pydantic_init_subclass__(
            *args,
            **{k: v for k, v in kwargs.items() if k not in TrackingGroup.model_fields},
        )

        # Initialize the tracking group
        cls.__DYNAPYDANTIC__: ty.ClassVar[TrackingGroup] = _init_tracking_group(
            cls, **kwargs
        )

        # Initialize our SubclassTrackingModel-specific config
        cls.__DYNAPYDANTIC_STM_CONFIG__: ty.ClassVar[_StmConfig] = _StmConfig.create(
            cls,
            exclude_from_union=exclude_from_union,
            union_realization=union_realization,
            inherited=getattr(cls, "__DYNAPYDANTIC_STM_CONFIG__", None),
        )

        # If we are going to be tracked, walk the entire MRO (to support
        # multi-level tree) and register ourselves with each oe.
        if not cls.__DYNAPYDANTIC_STM_CONFIG__.exclude_from_union:
            for base in cls.__mro__:
                if (
                    issubclass(base, SubclassTrackingModel)
                    and base is not SubclassTrackingModel
                ):
                    base.__DYNAPYDANTIC__.register_model(cls)

__init_subclass__(*args, **kwargs) -> None

Subclass hook

Source code in src/dynapydantic/subclass_tracking_model.py
38
39
40
41
42
43
44
45
46
47
48
49
50
def __init_subclass__(cls, *args, **kwargs) -> None:
    """Subclass hook"""
    # Intercept any kwargs that are intended for TrackingGroup or
    # __pydantic_init_subclass__
    sig = inspect.signature(SubclassTrackingModel.__pydantic_init_subclass__)
    super().__init_subclass__(
        *args,
        **{
            k: v
            for k, v in kwargs.items()
            if k not in TrackingGroup.model_fields and k not in sig.parameters
        },
    )

__pydantic_init_subclass__(*args, exclude_from_union: bool | None = None, union_realization: str | UnionRealization | None = None, **kwargs) -> None classmethod

Pydantic subclass hook

Source code in src/dynapydantic/subclass_tracking_model.py
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
@classmethod
def __pydantic_init_subclass__(
    cls,
    *args,
    exclude_from_union: bool | None = None,
    union_realization: str | UnionRealization | None = None,
    **kwargs,
) -> None:
    """Pydantic subclass hook"""
    # Forward along any unexpected arguments that were not intended
    # for TrackingGroup.
    super().__pydantic_init_subclass__(
        *args,
        **{k: v for k, v in kwargs.items() if k not in TrackingGroup.model_fields},
    )

    # Initialize the tracking group
    cls.__DYNAPYDANTIC__: ty.ClassVar[TrackingGroup] = _init_tracking_group(
        cls, **kwargs
    )

    # Initialize our SubclassTrackingModel-specific config
    cls.__DYNAPYDANTIC_STM_CONFIG__: ty.ClassVar[_StmConfig] = _StmConfig.create(
        cls,
        exclude_from_union=exclude_from_union,
        union_realization=union_realization,
        inherited=getattr(cls, "__DYNAPYDANTIC_STM_CONFIG__", None),
    )

    # If we are going to be tracked, walk the entire MRO (to support
    # multi-level tree) and register ourselves with each oe.
    if not cls.__DYNAPYDANTIC_STM_CONFIG__.exclude_from_union:
        for base in cls.__mro__:
            if (
                issubclass(base, SubclassTrackingModel)
                and base is not SubclassTrackingModel
            ):
                base.__DYNAPYDANTIC__.register_model(cls)

TrackingGroup pydantic-model

Bases: BaseModel

Tracker for pydantic models

Fields:

Validators:

  • _ensure_union_mode
  • _coerce_union_mode
Source code in src/dynapydantic/tracking_group.py
 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
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
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."
        ),
    )
    union_mode: UnionMode | None = pydantic.Field(
        None,
        description=(
            "Union validation strategy. Pass a DiscriminatedConfig instance "
            'or one of the plain strings "smart" or "left_to_right". You can '
            "also just pass the fields for DiscriminatedConfig to this "
            "model and they will be forwarded."
        ),
    )
    discriminator_field: str | None = pydantic.Field(
        None,
        description=(
            "Name of the discriminator field. NOTE: This field is "
            "here as an alias for union_mode.discriminator_field. Passing "
            "both a discriminator_field and a union_mode will result in an "
            "error."
        ),
    )
    discriminator_value_generator: ty.Callable[[type], str] | None = pydantic.Field(
        None,
        description=(
            "A callable that produces default values for 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."
        ),
    )
    models: dict[str, type[pydantic.BaseModel]] = pydantic.Field(
        {},
        description="The tracked models",
    )

    _generation: int = pydantic.PrivateAttr(default=0)
    _adapter: pydantic.TypeAdapter | None = pydantic.PrivateAttr(default=None)
    _adapter_generation: int = pydantic.PrivateAttr(default=-1)

    @pydantic.model_validator(mode="after")
    def _ensure_union_mode(self) -> "TrackingGroup":
        """There must be a union_mode

        This validator works as a guard on _coerce_union_mode to make
        """
        if self.union_mode is None:
            msg = (
                "union_mode is required. This normally indicates that you "
                "subclasses TrackingGroup and wrote an invalid validator, but "
                "could also be a bug with dynapydantic, so please file a bug "
                "report with a reproducer on how you got here if you suspect "
                "a bug."
            )
            raise ValueError(msg)

        # Ensure the top-level fields are in-sync
        if isinstance(self.union_mode, DiscriminatedConfig):
            self.discriminator_field = self.union_mode.discriminator_field
            self.discriminator_value_generator = (
                self.union_mode.discriminator_value_generator
            )
        else:
            self.discriminator_field = None
            self.discriminator_value_generator = None

        return self

    @pydantic.model_validator(mode="before")
    @classmethod
    def _coerce_union_mode(cls, data: ty.Any) -> ty.Any:  # noqa: ANN401
        """Coerce flat discriminator kwargs into a DiscriminatedConfig.

        Allows callers to pass ``discriminator_field`` and
        ``discriminator_value_generator`` at the top level and transparently
        assembles a ``DiscriminatedConfig`` from them. This avoids an extra
        import/nesting layer for the user.
        """
        if not isinstance(data, dict):
            return data

        disc_field = data.get("discriminator_field", None)
        has_disc_field = disc_field is not None
        union_mode = data.get("union_mode", None)
        has_union_mode = union_mode is not None

        # If the user passed us both a discriminator field and a union_mode,
        # things must be perfectly consistent
        if has_disc_field and has_union_mode:
            consistent = (
                isinstance(union_mode, DiscriminatedConfig)
                and disc_field == union_mode.discriminator_field
                and data.get("discriminator_value_generator")
                is union_mode.discriminator_value_generator
            ) or (
                isinstance(union_mode, dict)
                and disc_field == union_mode.get("discriminator_field")
                and data.get("discriminator_value_generator")
                is union_mode.get("discriminator_value_generator")
            )
            if not consistent:
                msg = (
                    "Received both union_mode and discriminator_field; pass one "
                    "or the other."
                )
                raise ValueError(msg)

        if has_disc_field and not has_union_mode:
            # Forward arguments to DiscriminatedConfig
            data["union_mode"] = {
                "discriminator_field": disc_field,
                "discriminator_value_generator": data.get(
                    "discriminator_value_generator",
                ),
            }
        elif not has_disc_field and not has_union_mode:
            msg = "Either union_mode or discriminator_field must be given"
            raise ValueError(msg)

        return data

    @property
    def _discriminated(self) -> DiscriminatedConfig | None:
        """Return the DiscriminatedMode config, or None if not discriminated."""
        return (
            self.union_mode
            if isinstance(self.union_mode, DiscriminatedConfig)
            else None
        )

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

    @ty.overload
    def register(self, value: str | None = None) -> ty.Callable[[type], type]: ...

    @ty.overload
    def register(self, value: type[pydantic.BaseModel]) -> type[pydantic.BaseModel]: ...

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

        Parameters
        ----------
        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. Can also be the type
            itself to register (if the ()'s are omitted from the decorator).
        """
        if isinstance(value, type):
            self.register_model(value)
            return value

        def _wrapper(cls: type[pydantic.BaseModel]) -> type[pydantic.BaseModel]:
            self.register_model(cls, value)
            return cls

        return _wrapper

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

        Parameters
        ----------
        cls
            The model to register
        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.
        """
        if discriminator_value is not None and not isinstance(discriminator_value, str):
            msg = (
                "discriminator_value must be a str if given, was "
                f"{type(discriminator_value).__name__}"
            )
            raise RegistrationError(msg)

        if not isinstance(cls, type) or not issubclass(cls, pydantic.BaseModel):
            msg = (
                "only pydantic BaseModel subclasses can be registered in a "
                f"TrackingGroup. Got {cls}, which was not."
            )
            raise RegistrationError(msg)

        if (dm := self._discriminated) is not None:
            disc = dm.discriminator_field
            field = cls.model_fields.get(disc)

            if field is None:
                if discriminator_value is not None:
                    _inject_discriminator_field(cls, disc, discriminator_value)
                elif dm.discriminator_value_generator is not None:
                    _inject_discriminator_field(
                        cls,
                        disc,
                        dm.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, discriminator_value_generator "
                        f'was None and the "{disc}" field was not defined.'
                    )
                    raise RegistrationError(msg)
            elif ty.get_origin(field.annotation) is not ty.Literal:
                msg = (
                    f'the discriminator field "{disc}" already existed in '
                    f"{cls.__name__}, but its type annotation was "
                    f"{field.annotation}, not Literal."
                )
                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, the passed value was "{discriminator_value}" '
                    f' and "{field.default}" via the discriminator '
                    f"field ({disc})."
                )
                raise AmbiguousDiscriminatorValueError(msg)

            self._register_with_discriminator_field(cls)
        else:
            if discriminator_value is not None:
                warnings.warn(
                    f'A discriminator_value of "{discriminator_value}" was '
                    f"explicitly passed for {cls.__name__}, but "
                    f'union_mode="{self.union_mode}" does not use a '
                    "discriminator. The value will be ignored.",
                    stacklevel=2,
                )
            self._register_plain(cls)

    def union(
        self,
        *,
        plain: bool | None = None,
    ) -> ty.Any:  # noqa: ANN401
        """Return the union of all registered models

        Parameters
        ----------
        plain
            If set to `True`, a plain union of all members will be returned.
            Otherwise, the returned union will be annotated in accordance with
            the union mode.

        Returns
        -------
        Any
            If there is 1 registered type, the type itself. If there is > 1, a
            union of all registered types. This union may be annotated if
            `plain` is not `True`.

        Raises
        ------
        NoRegisteredTypesError
            If no types have been registered yet.
        """
        n = len(self.models)
        if n == 0:
            msg = (
                "Unable to produce a union from the tracking group "
                f'"{self.name}", as no types have been registered yet.'
            )
            raise NoRegisteredTypesError(msg)
        if n == 1:
            return next(iter(self.models.values()))

        union_mode = "smart" if plain else self.union_mode

        if isinstance(union_mode, DiscriminatedConfig):
            return ty.Annotated[
                functools.reduce(
                    operator.or_,
                    tuple(
                        ty.Annotated[x, pydantic.Tag(v)] for v, x in self.models.items()
                    ),
                ),
                pydantic.Field(discriminator=union_mode.discriminator_field),
            ]

        plain_union = functools.reduce(operator.or_, self.models.values())
        if union_mode == "left_to_right":
            return ty.Annotated[plain_union, pydantic.Field(union_mode="left_to_right")]

        # "smart" mode is pydantic's default behavior on a plain union
        return plain_union

    @property
    def generation(self) -> int:
        """The generation of the tracking group.

        This is a counter that increments every time a new registration occurs
        """
        return self._generation

    @property
    def type_adapter(self) -> pydantic.TypeAdapter:
        """Get the pydantic TypeAdapter for the union of all group members"""
        if self.generation != self._adapter_generation:
            self._adapter = pydantic.TypeAdapter(self.union())
            self._adapter_generation = self.generation

        # casting because the if statement ensures it is non-None (because
        # _adapter_generation starts at -1 and generation increments from 0.
        return ty.cast("pydantic.TypeAdapter", self._adapter)

    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 = ty.cast("DiscriminatedConfig", self.union_mode).discriminator_field
        value = cls.model_fields[disc].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 not isinstance(value, str):
            msg = (
                f"{cls.__name__}.{disc} had a default value of {value}, which "
                f"was of type {type(value).__name__}, not str."
            )
            raise RegistrationError(msg)

        self._do_register(value, cls)

    def _register_plain(self, cls: type[pydantic.BaseModel]) -> None:
        """Register the model keyed by its class name.

        Used for smart / left_to_right modes where no discriminator field
        is involved.

        Parameters
        ----------
        cls
            The model to register.
        """
        self._do_register(str(id(cls)), cls)

    def _do_register(self, key: str, cls: type[pydantic.BaseModel]) -> None:
        """Register the given model under the given key

        Parameters
        ----------
        key
            The key under which to register the model
        cls
            The model to register.
        """
        if (other := self.models.get(key)) is not None:
            if other is not cls:
                msg = (
                    f'Cannot register {cls.__name__} under the "{key}" '
                    f"identifier, which is already in use by {other.__name__}."
                )
                raise RegistrationError(msg)
        else:
            self._generation += 1
            self.models[key] = cls

name: str 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.

union_mode: UnionMode | None = None pydantic-field

Union validation strategy. Pass a DiscriminatedConfig instance or one of the plain strings "smart" or "left_to_right". You can also just pass the fields for DiscriminatedConfig to this model and they will be forwarded.

discriminator_field: str | None = None pydantic-field

Name of the discriminator field. NOTE: This field is here as an alias for union_mode.discriminator_field. Passing both a discriminator_field and a union_mode will result in an error.

discriminator_value_generator: ty.Callable[[type], str] | None = None pydantic-field

A callable that produces default values for the discriminator field

plugin_entry_point: str | None = 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.

models: dict[str, type[pydantic.BaseModel]] = {} pydantic-field

The tracked models

generation: int property

The generation of the tracking group.

This is a counter that increments every time a new registration occurs

type_adapter: pydantic.TypeAdapter property

Get the pydantic TypeAdapter for the union of all group members

load_plugins() -> None

Load plugins to discover/register additional models

Source code in src/dynapydantic/tracking_group.py
200
201
202
203
204
205
206
207
208
209
210
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(value: str | type[pydantic.BaseModel] | None = None) -> ty.Callable[[type], type] | type[pydantic.BaseModel]

register(
    value: str | None = None,
) -> ty.Callable[[type], type]
register(
    value: type[pydantic.BaseModel],
) -> type[pydantic.BaseModel]

Register a model into this group (decorator)

Parameters:

Name Type Description Default
value str | type[BaseModel] | 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. Can also be the type itself to register (if the ()'s are omitted from the decorator).

None
Source code in src/dynapydantic/tracking_group.py
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
def register(
    self,
    value: str | type[pydantic.BaseModel] | None = None,
) -> ty.Callable[[type], type] | type[pydantic.BaseModel]:
    """Register a model into this group (decorator)

    Parameters
    ----------
    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. Can also be the type
        itself to register (if the ()'s are omitted from the decorator).
    """
    if isinstance(value, type):
        self.register_model(value)
        return value

    def _wrapper(cls: type[pydantic.BaseModel]) -> type[pydantic.BaseModel]:
        self.register_model(cls, value)
        return cls

    return _wrapper

register_model(cls: type[pydantic.BaseModel], discriminator_value: str | None = None) -> None

Register the given model into this group

Parameters:

Name Type Description Default
cls type[BaseModel]

The model to register

required
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
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
def register_model(
    self,
    cls: type[pydantic.BaseModel],
    discriminator_value: str | None = None,
) -> None:
    """Register the given model into this group

    Parameters
    ----------
    cls
        The model to register
    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.
    """
    if discriminator_value is not None and not isinstance(discriminator_value, str):
        msg = (
            "discriminator_value must be a str if given, was "
            f"{type(discriminator_value).__name__}"
        )
        raise RegistrationError(msg)

    if not isinstance(cls, type) or not issubclass(cls, pydantic.BaseModel):
        msg = (
            "only pydantic BaseModel subclasses can be registered in a "
            f"TrackingGroup. Got {cls}, which was not."
        )
        raise RegistrationError(msg)

    if (dm := self._discriminated) is not None:
        disc = dm.discriminator_field
        field = cls.model_fields.get(disc)

        if field is None:
            if discriminator_value is not None:
                _inject_discriminator_field(cls, disc, discriminator_value)
            elif dm.discriminator_value_generator is not None:
                _inject_discriminator_field(
                    cls,
                    disc,
                    dm.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, discriminator_value_generator "
                    f'was None and the "{disc}" field was not defined.'
                )
                raise RegistrationError(msg)
        elif ty.get_origin(field.annotation) is not ty.Literal:
            msg = (
                f'the discriminator field "{disc}" already existed in '
                f"{cls.__name__}, but its type annotation was "
                f"{field.annotation}, not Literal."
            )
            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, the passed value was "{discriminator_value}" '
                f' and "{field.default}" via the discriminator '
                f"field ({disc})."
            )
            raise AmbiguousDiscriminatorValueError(msg)

        self._register_with_discriminator_field(cls)
    else:
        if discriminator_value is not None:
            warnings.warn(
                f'A discriminator_value of "{discriminator_value}" was '
                f"explicitly passed for {cls.__name__}, but "
                f'union_mode="{self.union_mode}" does not use a '
                "discriminator. The value will be ignored.",
                stacklevel=2,
            )
        self._register_plain(cls)

union(*, plain: bool | None = None) -> ty.Any

Return the union of all registered models

Parameters:

Name Type Description Default
plain bool | None

If set to True, a plain union of all members will be returned. Otherwise, the returned union will be annotated in accordance with the union mode.

None

Returns:

Type Description
Any

If there is 1 registered type, the type itself. If there is > 1, a union of all registered types. This union may be annotated if plain is not True.

Raises:

Type Description
NoRegisteredTypesError

If no types have been registered yet.

Source code in src/dynapydantic/tracking_group.py
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
def union(
    self,
    *,
    plain: bool | None = None,
) -> ty.Any:  # noqa: ANN401
    """Return the union of all registered models

    Parameters
    ----------
    plain
        If set to `True`, a plain union of all members will be returned.
        Otherwise, the returned union will be annotated in accordance with
        the union mode.

    Returns
    -------
    Any
        If there is 1 registered type, the type itself. If there is > 1, a
        union of all registered types. This union may be annotated if
        `plain` is not `True`.

    Raises
    ------
    NoRegisteredTypesError
        If no types have been registered yet.
    """
    n = len(self.models)
    if n == 0:
        msg = (
            "Unable to produce a union from the tracking group "
            f'"{self.name}", as no types have been registered yet.'
        )
        raise NoRegisteredTypesError(msg)
    if n == 1:
        return next(iter(self.models.values()))

    union_mode = "smart" if plain else self.union_mode

    if isinstance(union_mode, DiscriminatedConfig):
        return ty.Annotated[
            functools.reduce(
                operator.or_,
                tuple(
                    ty.Annotated[x, pydantic.Tag(v)] for v, x in self.models.items()
                ),
            ),
            pydantic.Field(discriminator=union_mode.discriminator_field),
        ]

    plain_union = functools.reduce(operator.or_, self.models.values())
    if union_mode == "left_to_right":
        return ty.Annotated[plain_union, pydantic.Field(union_mode="left_to_right")]

    # "smart" mode is pydantic's default behavior on a plain union
    return plain_union

DiscriminatedConfig pydantic-model

Bases: BaseModel

Configuration for a discriminated union.

Carries the discriminator field name and optional value generator that are required when pydantic's discriminated-union validation strategy is used.

Fields:

Source code in src/dynapydantic/union_mode.py
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class DiscriminatedConfig(pydantic.BaseModel, frozen=True, extra="forbid"):
    """Configuration for a discriminated union.

    Carries the discriminator field name and optional value generator that
    are required when pydantic's discriminated-union validation strategy is
    used.
    """

    discriminator_field: str = pydantic.Field(
        description="Name of the field used to discriminate between subtypes.",
    )
    discriminator_value_generator: Callable[[type], str] | None = pydantic.Field(
        None,
        description=(
            "A callable that produces default values for the discriminator "
            "field when none is supplied via register()."
        ),
    )

discriminator_field: str pydantic-field

Name of the field used to discriminate between subtypes.

discriminator_value_generator: Callable[[type], str] | None = None pydantic-field

A callable that produces default values for the discriminator field when none is supplied via register().

UnionRealization

Bases: Enum

When unions should be realized

Attributes:

Name Type Description
MODEL_CONSTRUCTION

Unions are realized at model construction time, during generation of the schema. This approach has the lowest runtime overhead, but is somewhat sensitive to ordering and may require model rebuilding for recursive model.

VALIDATION

Unions are realized at validation time. This approach is robust to order of operation, but carries additional runtime overhead.

Source code in src/dynapydantic/union_mode.py
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
class UnionRealization(enum.Enum):
    """When unions should be realized

    Attributes
    ----------
    MODEL_CONSTRUCTION
        Unions are realized at model construction time, during generation of
        the schema. This approach has the lowest runtime overhead, but is
        somewhat sensitive to ordering and may require model rebuilding for
        recursive model.
    VALIDATION
        Unions are realized at validation time. This approach is robust to order
        of operation, but carries additional runtime overhead.
    """

    MODEL_CONSTRUCTION = "model-construction"
    VALIDATION = "validation"

load_plugins(entity: TrackingGroup | type[SubclassTrackingModel]) -> None

Load plugins to discover/register additional models

Parameters:

Name Type Description Default
entity TrackingGroup | type[SubclassTrackingModel]

The entity for which to load plugins

required
Source code in src/dynapydantic/free_funcs.py
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def load_plugins(entity: TrackingGroup | type[SubclassTrackingModel]) -> None:
    """Load plugins to discover/register additional models

    Parameters
    ----------
    entity
        The entity for which to load plugins
    """
    if isinstance(entity, TrackingGroup):
        entity.load_plugins()
    elif isinstance(entity, type) and issubclass(entity, SubclassTrackingModel):
        entity.__DYNAPYDANTIC__.load_plugins()
    else:
        msg = (
            "dynapydantic.load_plugins() works on TrackingGroup or "
            f"SubclassTrackingModel, was given {entity}, which was neither."
        )
        raise TypeError(msg)

registered_models(entity: TrackingGroup | type[ModelT]) -> dict[str, type[pydantic.BaseModel]] | dict[str, type[ModelT]]

registered_models(
    entity: TrackingGroup,
) -> dict[str, type[pydantic.BaseModel]]
registered_models(
    entity: type[ModelT],
) -> dict[str, type[ModelT]]

Get the mapping of identifier -> model for all models tracked by the entity

Parameters:

Name Type Description Default
entity TrackingGroup | type[ModelT]

The entity for which to the registered models are desired

required

Returns:

Type Description
dict[str, BaseModel]

A mapping of identifier to the registered model. This identifier will be the discriminator value for entities that produce discriminated unions. If the entity produces a non-discriminated union, the identifier will just be some unique string.

Source code in src/dynapydantic/free_funcs.py
 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
def registered_models(
    entity: TrackingGroup | type[ModelT],
) -> dict[str, type[pydantic.BaseModel]] | dict[str, type[ModelT]]:
    """Get the mapping of identifier -> model for all models tracked by the entity

    Parameters
    ----------
    entity
        The entity for which to the registered models are desired

    Returns
    -------
    dict[str, pydantic.BaseModel]
        A mapping of identifier to the registered model. This identifier will be
        the discriminator value for entities that produce discriminated unions.
        If the entity produces a non-discriminated union, the identifier will
        just be some unique string.
    """
    if isinstance(entity, TrackingGroup):
        return entity.models
    if isinstance(entity, type) and issubclass(entity, SubclassTrackingModel):
        return entity.__DYNAPYDANTIC__.models

    msg = (
        "dynapydantic.registered_models() works on TrackingGroup or "
        f"SubclassTrackingModel, was given {entity}, which was neither."
    )
    raise TypeError(msg)

union(entity: TrackingGroup | type[SubclassTrackingModel], *, plain: bool | None = None) -> ty.Any

Get the union of all tracked models in this entity

Parameters:

Name Type Description Default
entity TrackingGroup | type[SubclassTrackingModel]

The entity for which the union shall be computed.

required
plain bool | None

If set to True, a plain union of all members will be returned. Otherwise, the returned union will be annotated in accordance with the union mode.

None

Returns:

Type Description
Any

If only 1 model is tracked, the model itself will be returned. If >1, model is tracked a union of all models will be returned. This union may be an Annotated union, depending on the union_mode of the entity and the value of plain.

See Also

Union[T] : Wrapper around this function for use in type annotations.

Raises:

Type Description
NoRegisteredTypesError

If no models are tracked by this entity.

Source code in src/dynapydantic/free_funcs.py
18
19
20
21
22
23
24
25
26
27
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
def union(
    entity: TrackingGroup | type[SubclassTrackingModel], *, plain: bool | None = None
) -> ty.Any:  # noqa: ANN401 (type is determined at runtime)
    """Get the union of all tracked models in this entity

    Parameters
    ----------
    entity
        The entity for which the union shall be computed.
    plain
        If set to `True`, a plain union of all members will be returned.
        Otherwise, the returned union will be annotated in accordance with
        the union mode.

    Returns
    -------
    Any
        If only 1 model is tracked, the model itself will be returned. If >1,
        model is tracked a union of all models will be returned. This union may
        be an `Annotated` union, depending on the `union_mode` of the entity and
        the value of `plain`.

    See Also
    --------
    [`Union[T]`][dynapydantic.Union] : Wrapper around this function for use
        in type annotations.

    Raises
    ------
    NoRegisteredTypesError
        If no models are tracked by this entity.
    """
    if isinstance(entity, TrackingGroup):
        return entity.union(plain=plain)
    if isinstance(entity, type) and issubclass(entity, SubclassTrackingModel):
        return entity.__DYNAPYDANTIC__.union(plain=plain)

    msg = (
        "dynapydantic.union() works on TrackingGroup or "
        f"SubclassTrackingModel, was given {entity}, which was neither."
    )
    raise TypeError(msg)