_util

Utilities for database migrations.

get_migration_defs(schema)

Extract all table and enum definitions from a schema module.

Source code in src/embar/db/_util.py

def get_migration_defs(schema: ModuleType) -> MigrationDefs:
    """
    Extract all table and enum definitions from a schema module.
    """
    enums: list[type[EnumBase]] = []
    tables: list[type[Table]] = []
    for name in dir(schema):
        obj = getattr(schema, name)
        # Check if it's a class and inherits from Table
        if isinstance(obj, type) and issubclass(obj, Table) and obj is not Table:
            tables.append(obj)
        if isinstance(obj, type) and issubclass(obj, EnumBase) and obj is not EnumBase:
            enums.append(obj)
    return MigrationDefs(enums=enums, tables=tables)

merge_ddls(defs)

Convert migration definitions to DDL statements in dependency order.

Source code in src/embar/db/_util.py

def merge_ddls(defs: MigrationDefs) -> list[Ddl]:
    """
    Convert migration definitions to DDL statements in dependency order.
    """
    queries: list[Ddl] = []
    for enum in defs.enums:
        queries.append(Ddl(name=enum.name, ddl=enum.ddl()))

    tables = _topological_sort_tables(defs.tables)
    for table in tables:
        constraints: list[str] = []
        for constraint in table.embar_config.constraints:
            constraints.append(constraint.sql().merged())
        queries.append(Ddl(name=table.get_name(), ddl=table.ddl(), constraints=constraints))

    return queries