arcanus.materia

The backend plugin layer. BaseMateria defines the contract every backend implements; NoOpMateria is the default, backend-less implementation.

arcanus.materia.base

BaseMateria

Source code in arcanus/materia/base.py

class BaseMateria:
    formulars: BidirectonDict[type[Transmuter], type[TransmuterProxied]]
    token: Optional[Token[BaseMateria]]
    column_type: type[Column[Any]]
    expression_compiler: ExpressionCompiler[Any, Any]

    def __init__(self) -> None:
        # Local import avoids a startup cycle with expression's active materia lookup.
        from arcanus.expression import Column, NativeExpressionCompiler

        self.formulars = BidirectonDict()
        self.token = None
        self.column_type = Column
        self.expression_compiler = NativeExpressionCompiler()

    @contextmanager
    def __call__(self):
        """Create a shallow copy of the materia and activate it as a context manager.

        The copy shares the validation context (formulars) with the original instance.

        Yields:
            A shallow copy set as the active materia.

        Example:
            materia = SqlalchemyMateria()
            with materia():
                instance = MyModel.model_validate(orm_obj)
        """
        copied = shallow_copy(self)

        copied.token = active_materia.set(copied)
        try:
            yield copied
        finally:
            active_materia.reset(copied.token)

    def __enter__(self) -> Self:
        if self.token is not None:
            raise RuntimeError(
                "This materia is already active. Use materia() to create "
                "a nested context: `with materia(): ...`"
            )
        self.token = active_materia.set(self)
        return self

    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        if self.token:
            active_materia.reset(self.token)
            self.token = None

    def __getitem__(
        self, transmuter: TransmuterMetaclass
    ) -> type[TransmuterProxied] | None:
        return self.formulars.get(transmuter)

    def __contains__(self, transmuter: TransmuterMetaclass) -> bool:
        return transmuter in self.formulars

    def bless(self, materia: Any):
        def decorator(transmuter_cls: type[T]) -> type[T]:
            # Check if materia implements TransmuterProxied by verifying required attributes
            if not hasattr(materia, "transmuter_proxy"):
                raise TypeError(
                    f"{self.__class__.__name__} require materia must implement TransmuterProxied."
                )

            # Prevent blessing the same provider to multiple transmuters
            existing = self.formulars.reverse.get(materia)
            if existing is not None and existing is not transmuter_cls:
                raise ValueError(
                    f"Provider {materia.__name__} is already blessed with "
                    f"transmuter {existing.__name__} in this materia. "
                    f"A provider cannot be blessed to multiple transmuters "
                    f"within a single materia."
                )

            self.formulars[transmuter_cls] = materia
            return transmuter_cls

        return decorator

    @staticmethod
    def transmuter_before_validator(
        transmuter_type: type[T], materia: M, info: ValidationInfo
    ) -> M:
        return materia

    @staticmethod
    def transmuter_after_validator(transmuter: T, info: ValidationInfo) -> T:
        return transmuter

    @staticmethod
    def transmuter_before_construct(transmuter_type: type[T], materia: object) -> dict:
        return materia.__dict__

    @staticmethod
    def transmuter_after_construct(transmuter: T) -> T:
        return transmuter

    @staticmethod
    def association_before_validator(
        association_type: type[A],
        value: Any,
        info: core_schema.ValidationInfo,
    ) -> Any:
        return value

    @staticmethod
    def association_after_validator(
        association: A, info: core_schema.ValidationInfo
    ) -> A:
        return association

    def load_association(self, association: Association) -> Any:
        raise NotImplementedError()

    async def aload_association(self, association: Association) -> Any:
        raise NotImplementedError()

    def association_loaded(self, association: Association) -> bool:
        """Whether reading *association* from its provider cannot fire a load.

        Materias with lazy loading (e.g. SQLAlchemy) override this; for
        everything else provider data is always in memory.
        """
        return True

association_loaded

association_loaded(association: Association) -> bool

Whether reading association from its provider cannot fire a load.

Materias with lazy loading (e.g. SQLAlchemy) override this; for everything else provider data is always in memory.

Source code in arcanus/materia/base.py

def association_loaded(self, association: Association) -> bool:
    """Whether reading *association* from its provider cannot fire a load.

    Materias with lazy loading (e.g. SQLAlchemy) override this; for
    everything else provider data is always in memory.
    """
    return True

JsonExpressionCompiler

Bases: ExpressionCompiler[object, str]

Compiles expressions to their neutral JSON form.

Backends with no native query layer (e.g. :class:NoOpMateria) use this, so expression() yields the same dict as expression.dump() instead of operating on a non-existent native column. The two stay in lock-step: each method mirrors the matching branch of Expression.dump / Order.dump.

Source code in arcanus/materia/base.py

class JsonExpressionCompiler(ExpressionCompiler[object, str]):
    """Compiles expressions to their neutral JSON form.

    Backends with no native query layer (e.g. :class:`NoOpMateria`) use this, so
    ``expression()`` yields the same dict as ``expression.dump()`` instead of
    operating on a non-existent native column. The two stay in lock-step: each
    method mirrors the matching branch of ``Expression.dump`` / ``Order.dump``.
    """

    def apply(self, column: Column[Any], operator: str, value: object) -> object:
        key = "in" if operator == "in_" else operator
        return {column.used_name: {key: dump(value)}}

    def and_(self, expressions: tuple[object, ...]) -> object:
        return {"and": list(expressions)}

    def or_(self, expressions: tuple[object, ...]) -> object:
        return {"or": list(expressions)}

    def not_(self, expression: object) -> object:
        return {"not": expression}

    def any_(self, column: Column[Any], value: Expression[bool] | None) -> object:
        return {column.used_name: dump(value)}

    def has(self, column: Column[Any], value: Expression[bool] | None) -> object:
        return {column.used_name: dump(value)}

    def order(self, order: Order[Any]) -> str:
        return order.dump()