Skip to content

Reflection

Reflection reads live database metadata through the native runtime and exposes an async inspector API for table names, columns, indexes, foreign keys, constraints, namespaces, reflected snapshots, model comparison, and model scaffolding.

ormdantic.reflection.Inspector 

Inspector(database)

Reflect live database metadata.

Source code in ormdantic/reflection.py 
def __init__(self, database: Any) -> None:
    self._database = database
    self._schema_cache: dict[_CacheKey, SchemaSnapshot] = {}

clear_cache class-attribute instance-attribute 

clear_cache = invalidate_cache

invalidate_cache 

invalidate_cache(
    *,
    schema=None,
    include_tables=None,
    exclude_tables=None,
    name_patterns=None
)

Invalidate cached reflection results.

Source code in ormdantic/reflection.py 
def invalidate_cache(
    self,
    *,
    schema: str | None = None,
    include_tables: Sequence[str] | None = None,
    exclude_tables: Sequence[str] | None = None,
    name_patterns: Sequence[str] | None = None,
) -> None:
    """Invalidate cached reflection results."""
    if (
        schema is None
        and include_tables is None
        and exclude_tables is None
        and name_patterns is None
    ):
        self._schema_cache.clear()
        return
    self._schema_cache.pop(
        self._cache_key(schema, include_tables, exclude_tables, name_patterns),
        None,
    )

schema async 

schema(
    *,
    schema=None,
    include_tables=None,
    exclude_tables=None,
    name_patterns=None,
    refresh=False
)

Return a reflected schema snapshot.

Source code in ormdantic/reflection.py 
async def schema(
    self,
    *,
    schema: str | None = None,
    include_tables: Sequence[str] | None = None,
    exclude_tables: Sequence[str] | None = None,
    name_patterns: Sequence[str] | None = None,
    refresh: bool = False,
) -> SchemaSnapshot:
    """Return a reflected schema snapshot."""
    return await self._reflect(
        "schema",
        None,
        lambda: self._schema(
            schema=schema,
            include_tables=include_tables,
            exclude_tables=exclude_tables,
            name_patterns=name_patterns,
            refresh=refresh,
        ),
    )

schema_dict async 

schema_dict(
    *,
    schema=None,
    include_tables=None,
    exclude_tables=None,
    name_patterns=None,
    refresh=False
)

Return a reflected schema snapshot as a dictionary.

Source code in ormdantic/reflection.py 
async def schema_dict(
    self,
    *,
    schema: str | None = None,
    include_tables: Sequence[str] | None = None,
    exclude_tables: Sequence[str] | None = None,
    name_patterns: Sequence[str] | None = None,
    refresh: bool = False,
) -> dict[str, Any]:
    """Return a reflected schema snapshot as a dictionary."""
    return await self._reflect(
        "schema_dict",
        None,
        lambda: self._schema(
            schema=schema,
            include_tables=include_tables,
            exclude_tables=exclude_tables,
            name_patterns=name_patterns,
            refresh=refresh,
        ).to_dict(),
    )

table_names async 

table_names(
    *,
    schema=None,
    include_tables=None,
    exclude_tables=None,
    name_patterns=None,
    refresh=False
)

Return table names for supported dialects.

Source code in ormdantic/reflection.py 
async def table_names(
    self,
    *,
    schema: str | None = None,
    include_tables: Sequence[str] | None = None,
    exclude_tables: Sequence[str] | None = None,
    name_patterns: Sequence[str] | None = None,
    refresh: bool = False,
) -> list[str]:
    """Return table names for supported dialects."""
    return await self._reflect(
        "table_names",
        None,
        lambda: [
            table.name
            for table in self._schema(
                schema=schema,
                include_tables=include_tables,
                exclude_tables=exclude_tables,
                name_patterns=name_patterns,
                refresh=refresh,
            ).tables
        ],
    )

tables async 

tables(
    *,
    schema=None,
    include_tables=None,
    exclude_tables=None,
    name_patterns=None,
    refresh=False
)

Return reflected table metadata.

Source code in ormdantic/reflection.py 
async def tables(
    self,
    *,
    schema: str | None = None,
    include_tables: Sequence[str] | None = None,
    exclude_tables: Sequence[str] | None = None,
    name_patterns: Sequence[str] | None = None,
    refresh: bool = False,
) -> list[dict[str, Any]]:
    """Return reflected table metadata."""
    return await self._reflect(
        "tables",
        None,
        lambda: [
            table.to_dict()
            for table in self._schema(
                schema=schema,
                include_tables=include_tables,
                exclude_tables=exclude_tables,
                name_patterns=name_patterns,
                refresh=refresh,
            ).tables
        ],
    )

namespaces async 

namespaces(
    *,
    schema=None,
    include_tables=None,
    exclude_tables=None,
    name_patterns=None,
    refresh=False
)

Return reflected namespace/schema metadata.

Source code in ormdantic/reflection.py 
async def namespaces(
    self,
    *,
    schema: str | None = None,
    include_tables: Sequence[str] | None = None,
    exclude_tables: Sequence[str] | None = None,
    name_patterns: Sequence[str] | None = None,
    refresh: bool = False,
) -> list[dict[str, Any]]:
    """Return reflected namespace/schema metadata."""
    return await self._reflect(
        "namespaces",
        None,
        lambda: [
            namespace.to_dict()
            for namespace in self._schema(
                schema=schema,
                include_tables=include_tables,
                exclude_tables=exclude_tables,
                name_patterns=name_patterns,
                refresh=refresh,
            ).namespaces
        ],
    )

schema_names async 

schema_names(
    *,
    schema=None,
    include_tables=None,
    exclude_tables=None,
    name_patterns=None,
    refresh=False
)

Return reflected namespace/schema names.

Source code in ormdantic/reflection.py 
async def schema_names(
    self,
    *,
    schema: str | None = None,
    include_tables: Sequence[str] | None = None,
    exclude_tables: Sequence[str] | None = None,
    name_patterns: Sequence[str] | None = None,
    refresh: bool = False,
) -> list[str]:
    """Return reflected namespace/schema names."""
    return await self._reflect(
        "schema_names",
        None,
        lambda: [
            namespace.name
            for namespace in self._schema(
                schema=schema,
                include_tables=include_tables,
                exclude_tables=exclude_tables,
                name_patterns=name_patterns,
                refresh=refresh,
            ).namespaces
        ],
    )

columns async 

columns(table, *, schema=None, refresh=False)

Return column metadata for a table.

Source code in ormdantic/reflection.py 
async def columns(
    self,
    table: str,
    *,
    schema: str | None = None,
    refresh: bool = False,
) -> list[dict[str, Any]]:
    """Return column metadata for a table."""
    return await self._reflect(
        "columns",
        table,
        lambda: [
            _column_metadata(column)
            for column in self._table(table, schema=schema, refresh=refresh).columns
        ],
    )

indexes async 

indexes(table, *, schema=None, refresh=False)

Return index metadata for a table.

Source code in ormdantic/reflection.py 
async def indexes(
    self,
    table: str,
    *,
    schema: str | None = None,
    refresh: bool = False,
) -> list[dict[str, Any]]:
    """Return index metadata for a table."""
    return await self._reflect(
        "indexes",
        table,
        lambda: [
            index.to_dict()
            for index in self._table(table, schema=schema, refresh=refresh).indexes
        ],
    )

foreign_keys async 

foreign_keys(table, *, schema=None, refresh=False)

Return foreign key metadata for a table.

Source code in ormdantic/reflection.py 
async def foreign_keys(
    self,
    table: str,
    *,
    schema: str | None = None,
    refresh: bool = False,
) -> list[dict[str, Any]]:
    """Return foreign key metadata for a table."""
    return await self._reflect(
        "foreign_keys",
        table,
        lambda: _foreign_key_metadata(
            self._table(table, schema=schema, refresh=refresh)
        ),
    )

unique_constraints async 

unique_constraints(table, *, schema=None, refresh=False)

Return UNIQUE constraint metadata for a table.

Source code in ormdantic/reflection.py 
async def unique_constraints(
    self,
    table: str,
    *,
    schema: str | None = None,
    refresh: bool = False,
) -> list[dict[str, Any]]:
    """Return UNIQUE constraint metadata for a table."""
    return await self._reflect(
        "unique_constraints",
        table,
        lambda: [
            constraint
            for constraint in _constraint_metadata(
                self._table(table, schema=schema, refresh=refresh)
            )
            if constraint["type"] == "unique"
        ],
    )

check_constraints async 

check_constraints(table, *, schema=None, refresh=False)

Return CHECK constraint metadata for a table.

Source code in ormdantic/reflection.py 
async def check_constraints(
    self,
    table: str,
    *,
    schema: str | None = None,
    refresh: bool = False,
) -> list[dict[str, Any]]:
    """Return CHECK constraint metadata for a table."""
    return await self._reflect(
        "check_constraints",
        table,
        lambda: [
            constraint
            for constraint in _constraint_metadata(
                self._table(table, schema=schema, refresh=refresh)
            )
            if constraint["type"] == "check"
        ],
    )

constraints async 

constraints(table, *, schema=None, refresh=False)

Return primary key, unique, check, foreign key, and exclusion constraints.

Source code in ormdantic/reflection.py 
async def constraints(
    self,
    table: str,
    *,
    schema: str | None = None,
    refresh: bool = False,
) -> list[dict[str, Any]]:
    """Return primary key, unique, check, foreign key, and exclusion constraints."""
    return await self._reflect(
        "constraints",
        table,
        lambda: _constraint_metadata(
            self._table(table, schema=schema, refresh=refresh)
        ),
    )

compare_to_models async 

compare_to_models(
    *,
    schema=None,
    include_tables=None,
    exclude_tables=None,
    name_patterns=None,
    refresh=False
)

Compare the reflected schema with registered Ormdantic models.

Source code in ormdantic/reflection.py 
async def compare_to_models(
    self,
    *,
    schema: str | None = None,
    include_tables: Sequence[str] | None = None,
    exclude_tables: Sequence[str] | None = None,
    name_patterns: Sequence[str] | None = None,
    refresh: bool = False,
) -> SchemaDiff:
    """Compare the reflected schema with registered Ormdantic models."""
    return await self._reflect(
        "compare_to_models",
        None,
        lambda: self._compare_to_models(
            schema=schema,
            include_tables=include_tables,
            exclude_tables=exclude_tables,
            name_patterns=name_patterns,
            refresh=refresh,
        ),
    )

scaffold_models async 

scaffold_models(
    *,
    schema=None,
    include_tables=None,
    exclude_tables=None,
    name_patterns=None,
    database_variable="db",
    refresh=False
)

Generate Pydantic model scaffolding from reflected tables.

Source code in ormdantic/reflection.py 
async def scaffold_models(
    self,
    *,
    schema: str | None = None,
    include_tables: Sequence[str] | None = None,
    exclude_tables: Sequence[str] | None = None,
    name_patterns: Sequence[str] | None = None,
    database_variable: str = "db",
    refresh: bool = False,
) -> str:
    """Generate Pydantic model scaffolding from reflected tables."""
    return await self._reflect(
        "scaffold_models",
        None,
        lambda: _scaffold_models(
            self._schema(
                schema=schema,
                include_tables=include_tables,
                exclude_tables=exclude_tables,
                name_patterns=name_patterns,
                refresh=refresh,
            ),
            database_variable=database_variable,
        ),
    )

For an onboarding walkthrough, see Existing Databases. For direct migration workflows, use db.migrations.live_snapshot() and db.migrations.autogenerate().