Registry class

saffier.Registry

Registry(database, **kwargs)

The command center for the models being generated for Saffier.

PARAMETER	DESCRIPTION
database	TYPE: Database
**kwargs	TYPE: Any DEFAULT: {}

Source code in saffier/core/connection/registry.py

def __init__(self, database: Database, **kwargs: Any) -> None:
    self.database: Database = database
    self.models: Dict[str, Any] = {}
    self.reflected: Dict[str, Any] = {}
    self.db_schema = kwargs.get("schema", None)
    self.extra: Mapping[str, Type["Database"]] = kwargs.pop("extra", {})

    self.schema = Schema(registry=self)

    self._metadata = (
        sqlalchemy.MetaData(schema=self.db_schema)
        if self.db_schema is not None
        else sqlalchemy.MetaData()
    )

database instance-attribute

database = database

models instance-attribute

models = {}

reflected instance-attribute

reflected = {}

db_schema instance-attribute

db_schema = get('schema', None)

extra instance-attribute

extra = pop('extra', {})

schema instance-attribute

schema = Schema(registry=self)

_metadata instance-attribute

_metadata = MetaData(schema=db_schema) if db_schema is not None else MetaData()

metadata property writable

metadata

declarative_base cached property

declarative_base

engine property

engine

sync_engine property

sync_engine

create_all async

create_all()

Source code in saffier/core/connection/registry.py

async def create_all(self) -> None:
    if self.db_schema:
        await self.schema.create_schema(self.db_schema, True)
    async with Database(self.database, force_rollback=False) as database:
        async with database.transaction():
            await database.create_all(self.metadata)

drop_all async

drop_all()

Source code in saffier/core/connection/registry.py

async def drop_all(self) -> None:
    if self.db_schema:
        await self.schema.drop_schema(self.db_schema, True, True)
    async with Database(self.database, force_rollback=False) as database:
        async with database.transaction():
            await database.drop_all(self.metadata)