BaseDatabaseModel

Base class for database models -- persistence-layer data schema definitions that map domain aggregates to specific storage technologies. Database models are typically auto-generated but can be customized for adapter-specific optimizations.

Bases: Element, OptionsMixin

Base class for database models -- persistence-layer representations that map between domain aggregates/entities and database-specific schemas.

Database models are the bridge between the domain model and the storage layer. Each persistence adapter (SQLAlchemy, Elasticsearch, etc.) provides its own concrete subclass. Protean auto-generates a default database model for every aggregate; use @domain.model only when you need to customize the schema mapping (e.g. column names, indexes, JSON serialization).

Subclasses must implement from_entity() for converting domain objects to database records. A default to_entity() is provided that handles the common pattern of iterating attributes with referenced_as remapping; override it only when storage-specific logic is needed (e.g. Elasticsearch's meta.id extraction).

Meta Options

Option	Type	Description
part_of	type	The aggregate or entity this model maps. Required.
database	str	The database provider name (default: "default").
schema_name	str	Override the storage table/collection name.

derive_schema_name classmethod

derive_schema_name() -> str

Derive the storage table/collection name.

Uses meta_.schema_name if explicitly set, otherwise falls back to the schema name of the associated aggregate or entity.

RETURNS	DESCRIPTION
str	The resolved schema name. TYPE: str

Source code in src/protean/core/database_model.py

@classmethod
def derive_schema_name(cls) -> str:
    """Derive the storage table/collection name.

    Uses ``meta_.schema_name`` if explicitly set, otherwise falls back
    to the schema name of the associated aggregate or entity.

    Returns:
        str: The resolved schema name.
    """
    if hasattr(cls.meta_, "schema_name") and cls.meta_.schema_name:
        return cast(str, cls.meta_.schema_name)
    else:
        return cast(str, cls.meta_.part_of.meta_.schema_name)

from_entity abstractmethod classmethod

from_entity(entity: Any) -> Any

Convert a domain entity/aggregate into a database model instance.

Implementors should extract field values from the domain object and map them to the database-specific representation.

PARAMETER	DESCRIPTION
entity	The domain entity or aggregate to convert. TYPE: BaseEntity

RETURNS	DESCRIPTION
Any	The database model instance ready for persistence.

Source code in src/protean/core/database_model.py

@classmethod
@abstractmethod
def from_entity(cls, entity: Any) -> Any:
    """Convert a domain entity/aggregate into a database model instance.

    Implementors should extract field values from the domain object and
    map them to the database-specific representation.

    Args:
        entity (BaseEntity): The domain entity or aggregate to convert.

    Returns:
        The database model instance ready for persistence.
    """

to_entity classmethod

to_entity(item: Any) -> BaseEntity

Convert a database record back into a domain entity/aggregate.

Iterates entity attributes, applies referenced_as remapping, and reconstructs the domain object. Adapters with storage-specific reconstruction needs (e.g. Elasticsearch meta.id extraction) should override this method entirely.

Source code in src/protean/core/database_model.py

@classmethod
def to_entity(cls, item: Any) -> "BaseEntity":
    """Convert a database record back into a domain entity/aggregate.

    Iterates entity attributes, applies ``referenced_as`` remapping,
    and reconstructs the domain object.  Adapters with storage-specific
    reconstruction needs (e.g. Elasticsearch ``meta.id`` extraction)
    should override this method entirely.
    """
    item_dict: dict[str, Any] = {}
    for attr_name, attr_obj in attributes(cls.meta_.part_of).items():
        referenced_as = attr_obj.referenced_as
        if referenced_as:
            # ``field_name`` is always populated for a registered field
            # (set by ``Field.__set_name__``); the ``or attr_name`` fallback
            # only satisfies the ``str | None`` type and is never taken at
            # runtime for fields returned by ``attributes()``.
            field_name = attr_obj.field_name or attr_name
            item_dict[field_name] = cls._get_value(item, referenced_as)
        else:
            item_dict[attr_name] = cls._get_value(item, attr_name)
    return cast("BaseEntity", cls.meta_.part_of(**item_dict))

to_records classmethod

to_records(
    items: list[Any], fields: list[str]
) -> list[Record]

Build read-only Record objects from storage records for fields.

Unlike :meth:to_entity, this does not materialize domain entities: it reads only the selected attributes and returns inert :class:~protean.core.queryset.Record objects, skipping all validation and invariants. It is the field-selection counterpart used by :meth:QuerySet.only.

Field metadata is resolved once up front and reused across every record, so the per-row work is just value reads.

Adapters with storage-specific record access (e.g. Elasticsearch's meta.id identity extraction) should override this method.

:param items: The raw storage records returned by _filter. :param fields: Attribute (column) names to read from each record.

Source code in src/protean/core/database_model.py

@classmethod
def to_records(cls, items: list[Any], fields: list[str]) -> list[Record]:
    """Build read-only ``Record`` objects from storage records for ``fields``.

    Unlike :meth:`to_entity`, this does **not** materialize domain
    entities: it reads only the selected attributes and returns inert
    :class:`~protean.core.queryset.Record` objects, skipping all validation
    and invariants. It is the field-selection counterpart used by
    :meth:`QuerySet.only`.

    Field metadata is resolved once up front and reused across every
    record, so the per-row work is just value reads.

    Adapters with storage-specific record access (e.g. Elasticsearch's
    ``meta.id`` identity extraction) should override this method.

    :param items: The raw storage records returned by ``_filter``.
    :param fields: Attribute (column) names to read from each record.
    """
    attrs = attributes(cls.meta_.part_of)
    entity_name = cls.meta_.part_of.__name__
    # Records are keyed by the attribute name the caller selected (what
    # they passed to ``only()``); the value is read from the storage key,
    # honouring ``referenced_as``. Resolve both once for the batch.
    resolved = [
        (attr_name, attrs[attr_name].referenced_as or attr_name)
        for attr_name in fields
    ]
    return [
        Record(
            entity_name,
            {key: cls._get_value(item, source) for key, source in resolved},
        )
        for item in items
    ]