Skip to content

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 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
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

_get_engine cached property 

_get_engine

declarative_base cached property 

declarative_base

engine property 

engine

_get_sync_engine cached property 

_get_sync_engine

sync_engine property 

sync_engine

_get_database_url 

_get_database_url()
Source code in saffier/core/connection/registry.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def _get_database_url(self) -> str:
    url = self.database.url
    if not url.driver:
        if url.dialect in settings.postgres_dialects:
            url = url.replace(driver="asyncpg")
        elif url.dialect in settings.mysql_dialects:
            url = url.replace(driver="aiomysql")
        elif url.dialect in settings.sqlite_dialects:
            url = url.replace(driver="aiosqlite")
        elif url.dialect in settings.mssql_dialects:
            raise ImproperlyConfigured("Saffier does not support MSSQL at the moment.")
    elif url.driver in settings.mssql_drivers:  # type: ignore
        raise ImproperlyConfigured("Saffier does not support MSSQL at the moment.")
    return str(url)

create_all async 

create_all()
Source code in saffier/core/connection/registry.py 
90
91
92
93
94
95
96
async def create_all(self) -> None:
    if self.db_schema:
        await self.schema.create_schema(self.db_schema, True)
    async with self.database:
        async with self.engine.begin() as connection:
            await connection.run_sync(self.metadata.create_all)
    await self.engine.dispose()

drop_all async 

drop_all()
Source code in saffier/core/connection/registry.py 
 98
 99
100
101
102
103
104
async def drop_all(self) -> None:
    if self.db_schema:
        await self.schema.drop_schema(self.db_schema, True, True)
    async with self.database:
        async with self.engine.begin() as conn:
            await conn.run_sync(self.metadata.drop_all)
    await self.engine.dispose()