pydantic-magic 0.1.0

Subclass driven extensible validation for pydantic

pydantic-magic is an extension of pydantic which provides validation-time delegation to model subclasses through a new base model class MagicModel. Functionally, the behavior is similar to pydantic's discriminated union but as any imported subclass can be targeted, models can refer to only the expected base type while remaining fully extensible.

from math import pi

from pydantic_magic import MagicModel

class Shape(MagicModel, abstract=True):
    def area(self) -> float:
        raise NotImplementedError

class Circle(Shape):
    radius: float

    def area(self) -> float:
        return pi * self.radius ** 2

class Square(Shape):
    side: float

    def area(self) -> float:
        return self.side ** 2

print(Shape.model_validate({"type": "circle", "radius": 5.0}).area())

NamedMagicModel extends this idea further by allowing model instances to be named and referenced by name during validation. When validating nested MagicModel types, any NamedMagicModel instance has its name registered to the parent validation frame so any sibling models can refer to it without duplication or an additional transformation layer.

from pydantic_magic import MagicModel, NamedMagicModel

class Worker(NamedMagicModel, abstract=True):
    pass

class FastWorker(Worker):
    threads: int

class Pipeline(MagicModel):
    workers: list[Worker]
    primary: Worker

pipeline = Pipeline.model_validate({
    "workers": [{"type": "fast", "name": "my_worker", "threads": 4}],
    "primary": "my_worker",
})
assert pipeline.primary is pipeline.workers[0]

Additionally, when validating fields of the type dict[str, NamedMagicModel], a validation helper NamedMagicModel.meta_from_key can be used to inject the instance name, and optionally type, from the dictionary key.

from typing import Annotated

from pydantic_magic import MagicModel, NamedMagicModel

class Worker(NamedMagicModel, abstract=True):
    pass

class FastWorker(Worker):
    threads: int

class Pipeline(MagicModel):
    # "type:name" key injects both type discriminator and instance name, update_key writes back
    # instance name to validated dictionary.
    workers: Annotated[dict[str, Worker], Worker.meta_from_key(update_key=True)]
    primary: Worker

pipeline = Pipeline.model_validate({
    "workers": {"fast:my_worker": {"threads": 4}},
    "primary": "my_worker",
})
assert isinstance(pipeline.primary, FastWorker)
assert {"my_worker": pipeline.primary} == pipeline.workers