Skip to content

types

Pydantic configuration persistence for SQLAlchemy.

This module provides a custom SQLAlchemy TypeDecorator for automatically serializing / deserializing Pydantic models to / from JSON fields in the RDBMS. The purpose is to provide tight integration with the Eleanor Framework settings layer eleanor.settings.resource. Broadly, dynamic Pydantic settings should be used on fields that are not filtered upon since we do not want to tightly couple the Pydantic object with the RDBMS index and query logic.

Warning

There is a bug where this only works for the settings field on a table. Adding Pydantic configurations to additional fields will not work. See https://git.kube/ML/eleanor/issues/142 for additional details.

Classes

PydanticType 

PydanticType(
    pydantic_type: Type[BaseModel],
    *args: Any,
    **kwargs: Any
)

Bases: TypeDecorator

A custom SQLAlchemy TypeDecorator for automatically converting JSON database fields to Pydantic models and vice versa.

See: https://docs.sqlalchemy.org/en/20/core/custom_types.html#sqlalchemy.types.TypeDecorator

Initializes the decorator with a specific Pydantic model type.

Parameters:

  • pydantic_type (Type[BaseModel]) –

    The Pydantic model class.

  • *args (Any, default: () ) –

    Additional positional arguments passed to the parent constructor.

  • **kwargs (Any, default: {} ) –

    Additional keyword arguments passed to the parent constructor.

Attributes

cache_ok class-attribute instance-attribute 
cache_ok = True
impl class-attribute instance-attribute 
impl = JSON
pydantic_type instance-attribute 
pydantic_type = pydantic_type
python_type property 
python_type: Type[Any]

Return the Python type object expected to be returned by instances of this type, if known.

Basically, for those types which enforce a return type, or are known across the board to do such for all common DBAPIs (like int for example), will return that type.

If a return type is not defined, raises NotImplementedError.

Note that any type also accommodates NULL in SQL which means you can also get back None from any type in practice.

Functions

__repr__ 
__repr__()

Needed for Alembic to work properly, basically we return the repr of the impl attribute. This methodology also need to have user_module_prefix="sa." set in the Alembic env.py file.

copy 
copy(**kwargs: Any) -> PydanticType

Creates a copy of this TypeDecorator instance, used by SQLAlchemy’s copying mechanism.

Parameters:

  • **kwargs (Any, default: {} ) –

    Keyword arguments to be passed to the constructor of the copy.

Returns:

  • JSONAsPydantic ( PydanticType ) –

    A new instance of this class.

process_bind_param 
process_bind_param(
    value: Any, dialect: Dialect
) -> Optional[str]

Processes the value before sending it to the database, converting the Pydantic model instance to a dictionary.

Parameters:

  • value (Any) –

    The value to be processed.

  • dialect (Dialect) –

    The SQLAlchemy dialect in use.

Returns:

  • Optional[str] –

    Optional[dict]: The processed value as a dictionary suitable for JSON serialization, or None.

process_literal_param 
process_literal_param(
    value: Optional[BaseModel], dialect: Dialect
) -> str

Receive a literal parameter value to be rendered inline within a statement.

.. note::

This method is called during the **SQL compilation** phase of a
statement, when rendering a SQL string. Unlike other SQL
compilation methods, it is passed a specific Python value to be
rendered as a string. However it should not be confused with the
:meth:`_types.TypeDecorator.process_bind_param` method, which is
the more typical method that processes the actual value passed to a
particular parameter at statement execution time.

Custom subclasses of :class:_types.TypeDecorator should override this method to provide custom behaviors for incoming data values that are in the special case of being rendered as literals.

The returned string will be rendered into the output string.

process_result_value 
process_result_value(
    value: Any, dialect: Dialect
) -> BaseModel

Processes the value fetched from the database, converting the JSON data to an instance of the specified Pydantic model.

Parameters:

  • value (Any) –

    The JSON data fetched from the database.

  • dialect (Dialect) –

    The SQLAlchemy dialect in use.

Returns:

  • BaseModel ( BaseModel ) –

    An instance of the specified Pydantic model.