Skip to content

Registry class

edgy.Registry

Registry(database, **kwargs)

The command center for the models of Edgy.

PARAMETER DESCRIPTION
database

TYPE: Database

**kwargs

TYPE: Any DEFAULT: {}

Source code in edgy/core/connection/registry.py
21
22
23
24
25
26
27
28
29
30
31
32
33
34
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 edgy/core/connection/registry.py
46
47
48
49
50
51
52
53
54
55
56
57
58
59
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("Edgy does not support MSSQL at the moment.")
    elif url.driver in settings.mssql_drivers:
        raise ImproperlyConfigured("Edgy does not support MSSQL at the moment.")
    return str(url)

create_all async

create_all()
Source code in edgy/core/connection/registry.py
89
90
91
92
93
94
95
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 edgy/core/connection/registry.py
 97
 98
 99
100
101
102
103
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()