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
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 = (
        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 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
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)

init_models 

init_models(*, init_column_mappers=True, init_class_attrs=True)
PARAMETER DESCRIPTION
init_column_mappers

TYPE: bool DEFAULT: True

init_class_attrs

TYPE: bool DEFAULT: True

Source code in edgy/core/connection/registry.py 
87
88
89
90
91
92
def init_models(self, *, init_column_mappers: bool=True, init_class_attrs: bool=True) -> None:
    for model_class in self.models.values():
        model_class.meta.full_init(init_column_mappers=init_column_mappers, init_class_attrs=init_class_attrs)

    for model_class in self.reflected.values():
        model_class.meta.full_init(init_column_mappers=init_column_mappers, init_class_attrs=init_class_attrs)

invalidate_models 

invalidate_models(*, clear_class_attrs=True)
PARAMETER DESCRIPTION
clear_class_attrs

TYPE: bool DEFAULT: True

Source code in edgy/core/connection/registry.py 
94
95
96
97
98
def invalidate_models(self, *, clear_class_attrs: bool=True) -> None:
    for model_class in self.models.values():
        model_class.meta.invalidate(clear_class_attrs=clear_class_attrs)
    for model_class in self.reflected.values():
        model_class.meta.invalidate(clear_class_attrs=clear_class_attrs)

create_all async 

create_all()
Source code in edgy/core/connection/registry.py 
100
101
102
103
104
105
106
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 
108
109
110
111
112
113
114
115
116
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)
            # let's invalidate everything and recalculate. We don't want references.
            self.invalidate_models()
    await self.engine.dispose()