fix: сверка DBML — табличные UNIQUE и CHECK, действия FK, неизвестные классы типов

I-1: табличный UniqueConstraint ORM сверяется как уникальный индекс по колонкам
(флаг колонки unique=True — по-прежнему в свойствах колонки); любое CHECK в ORM —
«отсутствует в DBML».
I-2: у совпавших целей FK сравниваются ON DELETE / ON UPDATE (нет в DBML = NO ACTION),
DEFERRABLE и INITIALLY.
I-3: неизвестный класс типа отрисовывается repr с параметрами (VARCHAR(length=50),
Uuid(as_uuid=False)) и никогда не совпадает с именем типа DBML.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

api/scripts/dbml_parity.py

@@ -6,9 +6,11 @@
 - множество таблиц (`таблица: отсутствует в ORM|DBML`) и схема таблиц ORM (`dkb`);
 - колонки: состав (`таблица.колонка: отсутствует в ORM|DBML`), порядок общих колонок, тип + nullability одной строкой
   в нотации DBML (`ожидалось numeric(18,2) NULL / фактически numeric(18,4) NOT NULL`), PK, unique, default;
-- FK по колонке-ссылке (`таблица.колонка: FK: ожидалось родитель.колонка / фактически …`);
+- FK по колонке-ссылке (`таблица.колонка: FK: ожидалось родитель.колонка / фактически …`), у совпавших целей —
+  ON DELETE / ON UPDATE (нет в DBML = `NO ACTION`), DEFERRABLE и INITIALLY;
 - индексы по набору колонок: уникальность, partial-условие (note `partial UNIQUE WHERE <expr>`), `NULLS NOT DISTINCT`;
-  составной `[pk]` в Indexes DBML — это первичный ключ, а не индекс;
+  табличный `UniqueConstraint` ORM считается уникальным индексом; составной `[pk]` в Indexes DBML — это первичный ключ,
+  а не индекс; CHECK-ограничений в DBML нет — любое CHECK в ORM лишнее;
 - enum'ы: множество имён и упорядоченный список значений.
 
 Формат каждой строки — `<объект>: ожидалось X / фактически Y` или `<объект>: отсутствует в ORM|DBML` (функции
@@ -35,6 +37,7 @@ from sqlalchemy import (
     JSON,
     BigInteger,
     Boolean,
+    CheckConstraint,
     Column,
     Date,
     DateTime,
@@ -45,6 +48,7 @@ from sqlalchemy import (
     String,
     Table,
     Text,
+    UniqueConstraint,
 )
 from sqlalchemy import Enum as SAEnum
 from sqlalchemy.dialects.postgresql import JSONB, UUID
@@ -56,6 +60,7 @@ DBML_FILE_NAME = "data-model-v5.13.dbml"
 ORM = "ORM"
 DBML = "DBML"
 ABSENT = "нет"
+NO_ACTION = "NO ACTION"
 
 _NUMERIC_RE = re.compile(r"numeric\((\d+),\s*(\d+)\)")
 _VERSION_RE = re.compile(r"v(\d+\.\d+)")
@@ -130,7 +135,8 @@ def render_orm_type(orm_type: object) -> str:
     """SQL-тип ORM-колонки в нотации DBML (`numeric(18,2)`, `varchar`, `timestamptz`, имя enum, ...).
 
     Классы сравниваются точно: `SmallInteger` — подкласс `Integer`, `Text` — подкласс `String`, `Float` — подкласс
-    `Numeric`; подкласс не должен «проходить» за базовый тип.
+    `Numeric`; подкласс не должен «проходить» за базовый тип. Неизвестный класс отрисовывается как `repr` и всегда
+    расходится с DBML.
     """
     if isinstance(orm_type, SAEnum):
         return str(orm_type.name) if orm_type.schema == SCHEMA else f"{orm_type.schema}.{orm_type.name}"
@@ -139,7 +145,9 @@ def render_orm_type(orm_type: object) -> str:
     cls = type(orm_type)
     if cls in _PARAMETRIZED_TYPES:
         return _PARAMETRIZED_TYPES[cls](orm_type)
-    return _SIMPLE_TYPE_NAMES.get(cls, cls.__name__.lower())
+    # неизвестный класс (в т.ч. `VARCHAR(50)`, `Uuid(as_uuid=False)`, подклассы) — repr с параметрами: он начинается
+    # с имени класса в CamelCase и потому никогда не совпадёт с именем типа DBML; длина/флаги не теряются молча
+    return _SIMPLE_TYPE_NAMES.get(cls) or repr(orm_type)
 
 
 def render_dbml_type(dbml_type: object) -> str | None:
@@ -269,28 +277,59 @@ def _one_column_problems(label: str, orm_col: Column[Any], dbml_col: DbmlColumn,
 # --- FK --------------------------------------------------------------------------------------------------------
 
 
+def _fk_action(action: object) -> str:
+    """Действие FK в каноническом виде; отсутствие действия в DBML и ORM — это `NO ACTION` PostgreSQL."""
+    return " ".join(str(action).split()).upper() if action else NO_ACTION
+
+
+def _fk_initially(initially: object) -> str:
+    return str(initially).upper() if initially else ABSENT
+
+
 def _fk_problems(metadata: MetaData, dbml: PyDBML, names: Collection[str]) -> list[str]:
-    """FK по колонке-ссылке: множество целей `родитель.колонка` в DBML против ORM."""
+    """FK по колонке-ссылке: множество целей `родитель.колонка`, затем у совпавших целей — ON DELETE/ON UPDATE и
-    expected: dict[tuple[str, str], set[str]] = {}
+    DEFERRABLE/INITIALLY (в DBML их нет, значит в ORM их быть не должно)."""
+    expected: dict[tuple[str, str], dict[str, dict[str, str]]] = {}
     for ref in dbml.refs:
         # `>`: table1 ссылается на table2 (many-to-one); `<` — наоборот; `-` (one-to-one) читаем как `>`
         child, child_cols, parent, parent_cols = ref.table1, ref.col1, ref.table2, ref.col2
         if ref.type == "<":
             child, child_cols, parent, parent_cols = parent, parent_cols, child, child_cols
         if child.name in names:
+            props = {
+                "ON DELETE": _fk_action(ref.on_delete),
+                "ON UPDATE": _fk_action(ref.on_update),
+                "DEFERRABLE": _yes_no(False),
+                "INITIALLY": ABSENT,
+            }
             for child_col, parent_col in zip(child_cols, parent_cols, strict=True):
-                expected.setdefault((child.name, child_col.name), set()).add(f"{parent.name}.{parent_col.name}")
+                targets = expected.setdefault((child.name, child_col.name), {})
-    actual: dict[tuple[str, str], set[str]] = {}
+                targets[f"{parent.name}.{parent_col.name}"] = props
+    actual: dict[tuple[str, str], dict[str, dict[str, str]]] = {}
     for table in metadata.tables.values():
         if table.name in names:
             for fk in table.foreign_keys:
-                target = f"{fk.column.table.name}.{fk.column.name}"
+                # DDL строится по ForeignKeyConstraint; ForeignKey(ondelete=...) копирует туда же — читаем оба
-                actual.setdefault((table.name, fk.parent.name), set()).add(target)
+                constraint = fk.constraint
+                props = {
+                    "ON DELETE": _fk_action(getattr(constraint, "ondelete", None) or fk.ondelete),
+                    "ON UPDATE": _fk_action(getattr(constraint, "onupdate", None) or fk.onupdate),
+                    "DEFERRABLE": _yes_no(bool(getattr(constraint, "deferrable", None) or fk.deferrable)),
+                    "INITIALLY": _fk_initially(getattr(constraint, "initially", None) or fk.initially),
+                }
+                targets = actual.setdefault((table.name, fk.parent.name), {})
+                targets[f"{fk.column.table.name}.{fk.column.name}"] = props
     problems = []
     for key in sorted(expected.keys() | actual.keys()):
-        want, have = expected.get(key, set()), actual.get(key, set())
+        want, have = expected.get(key, {}), actual.get(key, {})
-        if want != have:
+        label = f"{key[0]}.{key[1]}: FK"
-            problems.append(mismatch(f"{key[0]}.{key[1]}: FK", _targets(want), _targets(have)))
+        if want.keys() != have.keys():
+            problems.append(mismatch(label, _targets(set(want)), _targets(set(have))))
+            continue
+        for target in sorted(want):
+            for prop, value in want[target].items():
+                if have[target][prop] != value:
+                    problems.append(mismatch(f"{label} {prop}", value, have[target][prop]))
     return problems
 
 
@@ -338,6 +377,13 @@ def _index_problems(orm_table: Table, dbml_table: DbmlTable) -> list[str]:
         options = orm_index.dialect_options["postgresql"]
         props = _index_props(bool(orm_index.unique), _norm_where(options["where"]), bool(options["nulls_not_distinct"]))
         actual.setdefault(tuple(c.name for c in orm_index.columns), []).append(props)
+    for constraint in orm_table.constraints:
+        # табличный UniqueConstraint сверяется как уникальный индекс по тем же колонкам; ограничение, созданное
+        # флагом колонки `unique=True` (`_column_flag`), уже сверено в свойствах колонки
+        if isinstance(constraint, UniqueConstraint) and not getattr(constraint, "_column_flag", False):
+            options = constraint.dialect_options["postgresql"]
+            props = _index_props(True, None, bool(options["nulls_not_distinct"]))
+            actual.setdefault(tuple(c.name for c in constraint.columns), []).append(props)
     problems = []
     for columns in sorted(expected.keys() | actual.keys()):
         label = f"{dbml_table.name}: индекс ({', '.join(columns)})"
@@ -351,6 +397,15 @@ def _index_problems(orm_table: Table, dbml_table: DbmlTable) -> list[str]:
     return problems
 
 
+def _check_problems(orm_table: Table) -> list[str]:
+    """CHECK-ограничения: в DBML их нет (типы и enum'ы — нативные), любое CHECK в ORM — лишнее."""
+    return [
+        missing(f"{orm_table.name}: CHECK {constraint.sqltext}", DBML)
+        for constraint in orm_table.constraints
+        if isinstance(constraint, CheckConstraint)
+    ]
+
+
 # --- точки входа -----------------------------------------------------------------------------------------------
 
 
@@ -369,6 +424,7 @@ def find_problems(metadata: MetaData, tables: Collection[str], dbml: PyDBML) ->
                 problems.append(mismatch(f"{name}: схема", SCHEMA, orm_tables[name].schema))
             problems += _column_problems(orm_tables[name], dbml_tables[name])
             problems += _index_problems(orm_tables[name], dbml_tables[name])
+            problems += _check_problems(orm_tables[name])
     return problems + _fk_problems(metadata, dbml, {n for n in tables if n in dbml_tables})
 
 

api/tests/test_dbml_parity.py

@@ -4,8 +4,10 @@
 import pytest
 from pydbml.classes import Expression
 from sqlalchemy import (
+    VARCHAR,
     BigInteger,
     Boolean,
+    CheckConstraint,
     Date,
     DateTime,
     DefaultClause,
@@ -16,10 +18,13 @@ from sqlalchemy import (
     SmallInteger,
     String,
     Text,
+    UniqueConstraint,
+    Uuid,
     func,
     text,
 )
 from sqlalchemy.dialects.postgresql import JSON, JSONB, UUID
+from sqlalchemy.dialects.postgresql import VARCHAR as PG_VARCHAR
 
 from src.models import Base
 from src.models.enums import pg_enum
@@ -98,12 +103,21 @@ def _dbml_enum(name: str) -> object:
         (String(), "text", False),
         (String(), "varchar", True),
         (DateTime(timezone=True), "timestamptz", True),
+        # неизвестные/производные классы не должны терять параметры и «проходить» за тип DBML
+        (VARCHAR(50), "varchar", False),
+        (PG_VARCHAR(10), "varchar", False),
+        (Uuid(as_uuid=False), "uuid", False),
     ],
 )
 def test_type_mapper(orm_type: object, dbml_type: str, ok: bool) -> None:
     assert (type_problem(orm_type, dbml_type) is None) is ok
 
 
+def test_unknown_orm_type_class_keeps_parameters_in_report() -> None:
+    assert type_problem(VARCHAR(50), "varchar") == "ожидалось varchar / фактически VARCHAR(length=50)"
+    assert type_problem(Uuid(as_uuid=False), "uuid") == "ожидалось uuid / фактически Uuid(as_uuid=False)"
+
+
 def test_unknown_dbml_type_is_reported_not_ignored() -> None:
     assert "не поддержан" in (type_problem(String(), "geometry") or "")
 
@@ -230,3 +244,42 @@ def test_now_default_forms_agree() -> None:
         == canonical_default_orm(DefaultClause(text("CURRENT_TIMESTAMP")))
     )
     assert expected != canonical_default_dbml(False)
+
+
+def test_table_level_unique_constraint_is_compared_as_unique_index() -> None:
+    """Табличный UniqueConstraint ORM, которого нет в DBML, — расхождение (флаг колонки `unique=True` — не он)."""
+    metadata = clone_metadata()
+    metadata.tables["dkb.regional_office"].append_constraint(UniqueConstraint("name", "is_active"))
+    assert find_problems(metadata, {"regional_office"}) == [
+        "regional_office: индекс (name, is_active): отсутствует в DBML"
+    ]
+    metadata.tables["dkb.permission"].append_constraint(UniqueConstraint("name"))
+    assert find_problems(metadata, {"permission"}) == ["permission: индекс (name): отсутствует в DBML"]
+
+
+def test_check_constraint_is_reported_missing_in_dbml() -> None:
+    metadata = clone_metadata()
+    metadata.tables["dkb.elimination_rule"].append_constraint(CheckConstraint("rule_no > 0", name="rule_no_positive"))
+    assert find_problems(metadata, {"elimination_rule"}) == ["elimination_rule: CHECK rule_no > 0: отсутствует в DBML"]
+
+
+def test_fk_actions_deferrable_and_initially_are_compared() -> None:
+    """В DBML у ссылок нет действий (= NO ACTION) и DEFERRABLE: лишний CASCADE в ORM обязан всплыть."""
+    label = "app_user.regional_office_id: FK"
+    metadata = clone_metadata()
+    fk = next(f for f in metadata.tables["dkb.app_user"].foreign_keys if f.parent.name == "regional_office_id")
+    fk.ondelete = "CASCADE"  # только на ForeignKey, ограничение не тронуто
+    assert find_problems(metadata, {"app_user"}) == [f"{label} ON DELETE: ожидалось NO ACTION / фактически CASCADE"]
+    fk.ondelete = None
+    constraint = fk.constraint
+    assert constraint is not None
+    constraint.onupdate = "set null"
+    constraint.deferrable = True
+    constraint.initially = "deferred"
+    assert find_problems(metadata, {"app_user"}) == [
+        f"{label} ON UPDATE: ожидалось NO ACTION / фактически SET NULL",
+        f"{label} DEFERRABLE: ожидалось нет / фактически да",
+        f"{label} INITIALLY: ожидалось нет / фактически DEFERRED",
+    ]
+    constraint.onupdate, constraint.deferrable, constraint.initially = "NO ACTION", None, None
+    assert find_problems(metadata, {"app_user"}) == []  # явный NO ACTION равен отсутствию действия