Module arti.types.pydantic

None

None

View Source

from __future__ import annotations

from typing import Any, Protocol

from pydantic import BaseModel

from pydantic.fields import ModelField

from pydantic.fields import UndefinedType as _PydanticUndefinedType

from arti.internal.type_hints import lenient_issubclass

from arti.types import Struct, Type, TypeAdapter, TypeSystem

from arti.types.python import python_type_system

pydantic_type_system = TypeSystem(key="pydantic", extends=(python_type_system,))

class _PostFieldConversionHook(Protocol):

    def __call__(self, type_: Type, *, name: str, required: bool) -> Type:

        raise NotImplementedError()

def get_post_field_conversion_hook(type_: Any) -> _PostFieldConversionHook:

    if hasattr(type_, "_pydantic_type_system_post_field_conversion_hook_"):

        return type_._pydantic_type_system_post_field_conversion_hook_  # type: ignore[no-any-return]

    return lambda type_, *, name, required: type_

@pydantic_type_system.register_adapter

class BaseModelAdapter(TypeAdapter):

    artigraph = Struct

    system = BaseModel

    @staticmethod

    def _field_to_artigraph(

        field: ModelField, *, hints: dict[str, Any], type_system: TypeSystem

    ) -> Type:

        subtype = type_system.to_artigraph(field.outer_type_, hints=hints)

        return get_post_field_conversion_hook(subtype)(

            subtype,

            name=field.name,

            required=(

                True if isinstance(field.required, _PydanticUndefinedType) else field.required

            ),

        )

    @classmethod

    def to_artigraph(

        cls, type_: type[BaseModel], *, hints: dict[str, Any], type_system: TypeSystem

    ) -> Type:

        return Struct(

            name=type_.__name__,

            fields={

                field.name: cls._field_to_artigraph(field, hints=hints, type_system=type_system)

                for field in type_.__fields__.values()

            },

        )

    @classmethod

    def matches_system(cls, type_: Any, *, hints: dict[str, Any]) -> bool:

        return lenient_issubclass(type_, cls.system)

    @classmethod

    def to_system(

        cls, type_: Type, *, hints: dict[str, Any], type_system: TypeSystem

    ) -> type[BaseModel]:

        assert isinstance(type_, Struct)

        return type(

            f"{type_.name}",

            (BaseModel,),

            {

                "__annotations__": {

                    k: type_system.to_system(v, hints=hints) for k, v in type_.fields.items()

                }

            },

        )

Variables

pydantic_type_system

Functions

get_post_field_conversion_hook

def get_post_field_conversion_hook(
    type_: 'Any'
) -> '_PostFieldConversionHook'

View Source

def get_post_field_conversion_hook(type_: Any) -> _PostFieldConversionHook:

    if hasattr(type_, "_pydantic_type_system_post_field_conversion_hook_"):

        return type_._pydantic_type_system_post_field_conversion_hook_  # type: ignore[no-any-return]

    return lambda type_, *, name, required: type_

Classes

BaseModelAdapter

class BaseModelAdapter(
    /,
    *args,
    **kwargs
)

View Source

@pydantic_type_system.register_adapter

class BaseModelAdapter(TypeAdapter):

    artigraph = Struct

    system = BaseModel

    @staticmethod

    def _field_to_artigraph(

        field: ModelField, *, hints: dict[str, Any], type_system: TypeSystem

    ) -> Type:

        subtype = type_system.to_artigraph(field.outer_type_, hints=hints)

        return get_post_field_conversion_hook(subtype)(

            subtype,

            name=field.name,

            required=(

                True if isinstance(field.required, _PydanticUndefinedType) else field.required

            ),

        )

    @classmethod

    def to_artigraph(

        cls, type_: type[BaseModel], *, hints: dict[str, Any], type_system: TypeSystem

    ) -> Type:

        return Struct(

            name=type_.__name__,

            fields={

                field.name: cls._field_to_artigraph(field, hints=hints, type_system=type_system)

                for field in type_.__fields__.values()

            },

        )

    @classmethod

    def matches_system(cls, type_: Any, *, hints: dict[str, Any]) -> bool:

        return lenient_issubclass(type_, cls.system)

    @classmethod

    def to_system(

        cls, type_: Type, *, hints: dict[str, Any], type_system: TypeSystem

    ) -> type[BaseModel]:

        assert isinstance(type_, Struct)

        return type(

            f"{type_.name}",

            (BaseModel,),

            {

                "__annotations__": {

                    k: type_system.to_system(v, hints=hints) for k, v in type_.fields.items()

                }

            },

        )

---

Ancestors (in MRO)

• arti.types.TypeAdapter

Class variables

artigraph

key

priority

system

Static methods

matches_artigraph

def matches_artigraph(
    type_: 'Type',
    *,
    hints: 'dict[str, Any]'
) -> 'bool'

View Source

    @classmethod

    def matches_artigraph(cls, type_: Type, *, hints: dict[str, Any]) -> bool:

        return isinstance(type_, cls.artigraph)

matches_system

def matches_system(
    type_: 'Any',
    *,
    hints: 'dict[str, Any]'
) -> 'bool'

View Source

    @classmethod

    def matches_system(cls, type_: Any, *, hints: dict[str, Any]) -> bool:

        return lenient_issubclass(type_, cls.system)

to_artigraph

def to_artigraph(
    type_: 'type[BaseModel]',
    *,
    hints: 'dict[str, Any]',
    type_system: 'TypeSystem'
) -> 'Type'

View Source

    @classmethod

    def to_artigraph(

        cls, type_: type[BaseModel], *, hints: dict[str, Any], type_system: TypeSystem

    ) -> Type:

        return Struct(

            name=type_.__name__,

            fields={

                field.name: cls._field_to_artigraph(field, hints=hints, type_system=type_system)

                for field in type_.__fields__.values()

            },

        )

to_system

def to_system(
    type_: 'Type',
    *,
    hints: 'dict[str, Any]',
    type_system: 'TypeSystem'
) -> 'type[BaseModel]'

View Source

    @classmethod

    def to_system(

        cls, type_: Type, *, hints: dict[str, Any], type_system: TypeSystem

    ) -> type[BaseModel]:

        assert isinstance(type_, Struct)

        return type(

            f"{type_.name}",

            (BaseModel,),

            {

                "__annotations__": {

                    k: type_system.to_system(v, hints=hints) for k, v in type_.fields.items()

                }

            },

        )