class DiffResolver:
"""Walk two DatabaseState trees and emit typed diffs.
Emission order is top-down: parent diffs appear before their children in
the returned list. Execution ordering for downstream stages is handled by
:class:`DiffDependencyResolver` via ``depends_on``.
"""
def diff(self, current: DatabaseState, desired: DatabaseState) -> list[Diff]:
"""Return the typed diff list to transform ``current`` into ``desired``."""
diffs: list[Diff] = []
self._diff_databases(current, desired, diffs)
return diffs
def _diff_databases(self, current: Databases, desired: Databases, out: list[Diff]) -> None:
"""Descend into every database present in either state.
Catalogs (databases) are assumed to exist outside dtm's control — see
``docs/decisions/migration-generation.md``. The walker therefore never
emits ``DatabaseAdded`` / ``DatabaseRemoved`` / ``DatabaseChanged``;
it only walks into the schemas/tables/columns inside each catalog.
"""
current_by_name = {db.name: db for db in current.databases}
desired_by_name = {db.name: db for db in desired.databases}
for name in sorted(current_by_name.keys() | desired_by_name.keys()):
current_db = current_by_name.get(name)
desired_db = desired_by_name.get(name)
# Use whichever Database is present as the parent reference for
# children — preferring `desired` since it represents the goal
# state used in downstream rendering.
parent_db = desired_db or current_db
assert parent_db is not None # name came from the union of the two maps
current_schemas = current_db.schemas if current_db else []
desired_schemas = desired_db.schemas if desired_db else []
self._diff_schemas(parent_db, current_schemas, desired_schemas, out)
def _diff_schemas(self, parent_db: Database, current: list[Schema], desired: list[Schema], out: list[Diff]) -> None:
current_by_name = {s.name: s for s in current}
desired_by_name = {s.name: s for s in desired}
for name in sorted(desired_by_name.keys() - current_by_name.keys()):
self._cascade_schema_added(desired_by_name[name], parent_db, out)
for name in sorted(current_by_name.keys() - desired_by_name.keys()):
self._cascade_schema_removed(current_by_name[name], parent_db, out)
for name in sorted(current_by_name.keys() & desired_by_name.keys()):
old, new = current_by_name[name], desired_by_name[name]
if _differ(old, new, _SCHEMA_EXCLUDE):
out.append(SchemaChanged(old=old, new=new, parent_db=parent_db))
self._diff_schema_contents(parent_db, new, old, new, out)
def _diff_schema_contents(
self,
parent_db: Database,
parent_schema: Schema,
current_schema: Schema,
desired_schema: Schema,
out: list[Diff],
) -> None:
"""Dispatch over peer object types within a schema.
Today only tables exist at this level; views, materialized views and
sequences extend this dispatch when ``cloe_metadata`` supports them.
"""
self._diff_tables(parent_db, parent_schema, current_schema.tables, desired_schema.tables, out)
def _diff_tables(
self,
parent_db: Database,
parent_schema: Schema,
current: list[Table],
desired: list[Table],
out: list[Diff],
) -> None:
current_by_name = {t.name: t for t in current}
desired_by_name = {t.name: t for t in desired}
for name in sorted(desired_by_name.keys() - current_by_name.keys()):
self._cascade_table_added(desired_by_name[name], parent_schema, parent_db, out)
for name in sorted(current_by_name.keys() - desired_by_name.keys()):
self._cascade_table_removed(current_by_name[name], parent_schema, parent_db, out)
for name in sorted(current_by_name.keys() & desired_by_name.keys()):
old, new = current_by_name[name], desired_by_name[name]
if _differ(old, new, _TABLE_EXCLUDE):
out.append(TableChanged(old=old, new=new, parent_schema=parent_schema, parent_db=parent_db))
self._diff_columns(parent_db, parent_schema, new, old.columns, new.columns, out)
self._diff_primary_key(parent_db, parent_schema, old, new, out)
def _diff_columns( # noqa: PLR0913
self,
parent_db: Database,
parent_schema: Schema,
parent_table: Table,
current: list[Column],
desired: list[Column],
out: list[Diff],
) -> None:
current_by_name = {c.name: c for c in current}
desired_by_name = {c.name: c for c in desired}
for name in sorted(desired_by_name.keys() - current_by_name.keys()):
out.append(
ColumnAdded(
column=desired_by_name[name],
parent_table=parent_table,
parent_schema=parent_schema,
parent_db=parent_db,
)
)
for name in sorted(current_by_name.keys() - desired_by_name.keys()):
out.append(
ColumnRemoved(
column=current_by_name[name],
parent_table=parent_table,
parent_schema=parent_schema,
parent_db=parent_db,
)
)
for name in sorted(current_by_name.keys() & desired_by_name.keys()):
old, new = current_by_name[name], desired_by_name[name]
if _differ(old, new, _COLUMN_EXCLUDE):
out.append(
ColumnChanged(
old=old,
new=new,
parent_table=parent_table,
parent_schema=parent_schema,
parent_db=parent_db,
)
)
def _diff_primary_key(
self,
parent_db: Database,
parent_schema: Schema,
old_table: Table,
new_table: Table,
out: list[Diff],
) -> None:
"""Emit a single PrimaryKeyChanged when the table's key-column set changes."""
old_pk = [c.name for c in old_table.columns if c.is_key]
new_pk = [c.name for c in new_table.columns if c.is_key]
if old_pk != new_pk:
out.append(
PrimaryKeyChanged(
old_key_columns=old_pk,
new_key_columns=new_pk,
parent_table=new_table,
parent_schema=parent_schema,
parent_db=parent_db,
)
)
# ── cascade helpers ──────────────────────────────────────────────────────
def _cascade_schema_added(self, schema: Schema, parent_db: Database, out: list[Diff]) -> None:
out.append(SchemaAdded(schema_obj=schema, parent_db=parent_db))
for table in schema.tables:
self._cascade_table_added(table, schema, parent_db, out)
def _cascade_schema_removed(self, schema: Schema, parent_db: Database, out: list[Diff]) -> None:
out.append(SchemaRemoved(schema_obj=schema, parent_db=parent_db))
for table in schema.tables:
self._cascade_table_removed(table, schema, parent_db, out)
def _cascade_table_added(self, table: Table, parent_schema: Schema, parent_db: Database, out: list[Diff]) -> None:
# CREATE TABLE renders columns inline, so no ColumnAdded cascade.
out.append(TableAdded(table=table, parent_schema=parent_schema, parent_db=parent_db))
def _cascade_table_removed(self, table: Table, parent_schema: Schema, parent_db: Database, out: list[Diff]) -> None:
# DROP TABLE removes columns implicitly, so no ColumnRemoved cascade.
out.append(TableRemoved(table=table, parent_schema=parent_schema, parent_db=parent_db))