Skip to main content

registry

Type registry for the versioned cache.

Discovers every cache/types/<name>/v<N>/ package and maps (name, version) to its canonical aliases (ORM, Record, store) plus its optional migration functions (upgrade, downgrade). Mirrors bitfount.steps.registry: discovery is automatic and lazy — the first accessor call scans the types/ tree and imports each version package.

A version package must advertise itself by re-exporting ORM (the SQLAlchemy mapped class), Record (the Pydantic model), and store (the CRUD module) from its __init__.py. These three are required.

A version package may also re-export an upgrade(op) function (and, for hygiene, a downgrade(op)) that migrates the previous version's physical table into this one — i.e. vN.upgrade takes v(N-1) -> vN. Both are optional: v1 has no predecessor, and any version with no physical schema change simply omits them (a None upgrade is a valid no-op hop). The bodies are Alembic op-API migrations autogenerated from the ORM by python -m bitfount.cache.tools.gen_migration; the ORM (schema.py) is the single authoring source of truth.

Module​

Functions​

get_downgrade​

def get_downgrade(    name: str, version: int = 1,) ‑> collections.abc.Callable[..., typing.Any] | None:

Return name's downgrade(op) for version (vN -> v(version-1)).

None when the version declares no physical schema change. Emitted for rollback/hygiene; the runtime driver only ever moves versions upward.

get_orm​

def get_orm(name: str, version: int = 1) ‑> type[sqlalchemy.orm.decl_api.DeclarativeBase]:

Return the SQLAlchemy mapped class for name at version.

get_store​

def get_store(name: str, version: int = 1) ‑> module:

Return the CRUD store module for name at version.

get_type_cls​

def get_type_cls(name: str, version: int = 1) ‑> type[pydantic.main.BaseModel]:

Return the Pydantic record class for name at version.

get_upgrade​

def get_upgrade(    name: str, version: int = 1,) ‑> collections.abc.Callable[..., typing.Any] | None:

Return name's inbound upgrade(op) for version (v(version-1) -> vN).

None when the version declares no physical schema change (a valid no-op hop) — and always for v1, which has no predecessor.

latest​

def latest(name: str) ‑> int:

Return the highest available version for name.

type_names​

def type_names() ‑> list[str]:

Return every discovered type name.

versions​

def versions(name: str) ‑> list[int]:

Return the sorted versions available for name.

Classes​

TypeEntry​

class TypeEntry(    orm: type[DeclarativeBase],    record_cls: type[BaseModel],    store: ModuleType,    upgrade: Callable[..., Any] | None,    downgrade: Callable[..., Any] | None,):

A single registered type version.

upgrade / downgrade are the version's inbound Alembic op-API migration functions (vN.upgrade takes v(N-1) -> vN). Both are None when the version declares no physical schema change (and always for v1, which has no predecessor).

Variables​

  • static orm : type[sqlalchemy.orm.decl_api.DeclarativeBase]
  • static record_cls : type[pydantic.main.BaseModel]
  • static store : module