9e53fc64be
DB_SCHEMA (по умолчанию dkb, валидация идентификатора) читают models/base.py (SCHEMA), db/db.py (search_path),
alembic/env.py и миграции 0001-0005 (SCHEMA и строковые FK f"{SCHEMA}.x.id" вместо литералов). В контуре -
OPENBAO__APP__DB_SCHEMA. Офлайн-SQL `alembic upgrade head --sql` при значении по умолчанию совпадает побайтно.
Тесты: валидация настройки, миграции и сессия приложения в схеме dkb_alt; .env.example/.env.test/README/CLAUDE.md.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EAKRDNzXtRrwmWz9z6PZu3
452 lines
24 KiB
Python
452 lines
24 KiB
Python
"""Сверка ORM-метаданных с DBML (design D-3) — единственная реализация: её вызывают CLI `check_dbml_vs_orm.py` и тесты.
|
||
|
||
Модуль не импортирует `src` (на вход приходят `MetaData` и реестр enum'ов) и не зависит от pytest.
|
||
|
||
Что сравнивается:
|
||
- множество таблиц (`таблица: отсутствует в ORM|DBML`) и схема таблиц ORM (`dkb`);
|
||
- колонки: состав (`таблица.колонка: отсутствует в ORM|DBML`), порядок общих колонок, тип + nullability одной строкой
|
||
в нотации DBML (`ожидалось numeric(18,2) NULL / фактически numeric(18,4) NOT NULL`), PK, unique, default;
|
||
- FK по колонке-ссылке (`таблица.колонка: FK: ожидалось родитель.колонка / фактически …`), у совпавших целей —
|
||
ON DELETE / ON UPDATE (нет в DBML = `NO ACTION`), DEFERRABLE и INITIALLY;
|
||
- индексы по набору колонок: уникальность, partial-условие (note `partial UNIQUE WHERE <expr>`), `NULLS NOT DISTINCT`;
|
||
табличный `UniqueConstraint` ORM считается уникальным индексом; составной `[pk]` в Indexes DBML — это первичный ключ,
|
||
а не индекс; CHECK-ограничений в DBML нет — любое CHECK в ORM лишнее;
|
||
- enum'ы: множество имён и упорядоченный список значений.
|
||
|
||
Формат каждой строки — `<объект>: ожидалось X / фактически Y` или `<объект>: отсутствует в ORM|DBML` (функции
|
||
`mismatch` / `missing`), чтобы тесты и человек читали один и тот же текст.
|
||
|
||
Нетривиальные места. DBML-default — python-литерал pydbml (bool/int/float/str) или `Expression` (обратные кавычки,
|
||
напр. `now()`); ORM server_default — `text(...)` или SQL-функция (`func.now()`). Обе стороны приводятся к строке:
|
||
bool -> `true`/`false`, число -> как есть, строка -> в одинарных кавычках, выражение -> текст в нижнем регистре.
|
||
"""
|
||
|
||
import enum
|
||
import os
|
||
import re
|
||
from collections.abc import Callable, Collection, Mapping
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from pydbml import PyDBML
|
||
from pydbml.classes import Column as DbmlColumn
|
||
from pydbml.classes import Enum as DbmlEnum
|
||
from pydbml.classes import Expression
|
||
from pydbml.classes import Table as DbmlTable
|
||
from sqlalchemy import (
|
||
JSON,
|
||
BigInteger,
|
||
Boolean,
|
||
CheckConstraint,
|
||
Column,
|
||
Date,
|
||
DateTime,
|
||
Integer,
|
||
MetaData,
|
||
Numeric,
|
||
SmallInteger,
|
||
String,
|
||
Table,
|
||
Text,
|
||
UniqueConstraint,
|
||
)
|
||
from sqlalchemy import Enum as SAEnum
|
||
from sqlalchemy.dialects.postgresql import JSONB, UUID
|
||
from sqlalchemy.sql.elements import TextClause
|
||
from sqlalchemy.sql.functions import FunctionElement
|
||
|
||
#: Эталонная схема модели. Имя схемы в БД конфигурируемо (`DB_SCHEMA`), CLI сверяет ORM, загруженную с этим значением.
|
||
SCHEMA = "dkb"
|
||
DBML_FILE_NAME = "data-model-v5.17.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+)")
|
||
_PARTIAL_RE = re.compile(r"partial\s+UNIQUE\s+WHERE\s+(.+?)(?:\s+—.*)?$", re.IGNORECASE | re.DOTALL)
|
||
_NULLS_NOT_DISTINCT_RE = re.compile(r"NULLS\s+NOT\s+DISTINCT", re.IGNORECASE)
|
||
_SYNONYMS = {"current_timestamp": "now()", "current_timestamp()": "now()"}
|
||
#: Типы DBML, которые сверка умеет сопоставлять (плюс `numeric(p,s)` и enum'ы); прочие — явное расхождение.
|
||
SUPPORTED_DBML_TYPES = frozenset(
|
||
{"uuid", "varchar", "text", "smallint", "int", "bigint", "boolean", "date", "timestamptz", "jsonb"}
|
||
)
|
||
|
||
|
||
# --- формат строк отчёта ---------------------------------------------------------------------------------------
|
||
|
||
|
||
def mismatch(obj: str, expected: object, actual: object) -> str:
|
||
"""Строка расхождения значения: `<объект>: ожидалось X / фактически Y`."""
|
||
return f"{obj}: ожидалось {expected} / фактически {actual}"
|
||
|
||
|
||
def missing(obj: str, where: str) -> str:
|
||
"""Строка отсутствия объекта: `<объект>: отсутствует в ORM|DBML`."""
|
||
return f"{obj}: отсутствует в {where}"
|
||
|
||
|
||
# --- поиск и чтение DBML ---------------------------------------------------------------------------------------
|
||
|
||
|
||
def default_dbml_path() -> Path:
|
||
"""Путь к DBML по умолчанию: каталог docs из `ARM_KA_DOCS`, иначе соседний `docs` рядом с `arm-core`."""
|
||
docs = os.environ.get("ARM_KA_DOCS")
|
||
docs_dir = Path(docs) if docs else Path(__file__).resolve().parents[3] / "docs"
|
||
return docs_dir / "model" / DBML_FILE_NAME
|
||
|
||
|
||
def read_dbml_version(path: Path) -> str | None:
|
||
"""Версия из первой строки шапки DBML (`// ДКБ · Модель данных v5.13: ...`) или `None`, если её там нет."""
|
||
with path.open(encoding="utf-8") as file:
|
||
first_line = file.readline()
|
||
found = _VERSION_RE.search(first_line)
|
||
return found.group(1) if found else None
|
||
|
||
|
||
def parse_dbml(path: Path) -> PyDBML:
|
||
"""Разобрать DBML-файл через pydbml."""
|
||
return PyDBML.parse_file(path)
|
||
|
||
|
||
# --- типы ------------------------------------------------------------------------------------------------------
|
||
|
||
|
||
#: Типы без параметров: точный класс SQLAlchemy -> имя в DBML.
|
||
_SIMPLE_TYPE_NAMES: dict[type, str] = {
|
||
Text: "text",
|
||
SmallInteger: "smallint",
|
||
Integer: "int",
|
||
BigInteger: "bigint",
|
||
Boolean: "boolean",
|
||
Date: "date",
|
||
JSONB: "jsonb",
|
||
JSON: "json",
|
||
}
|
||
#: Типы с параметрами: точный класс -> отрисовка с параметрами (длина, точность, часовой пояс).
|
||
_PARAMETRIZED_TYPES: dict[type, Callable[[Any], str]] = {
|
||
Numeric: lambda t: f"numeric({t.precision},{t.scale})",
|
||
String: lambda t: "varchar" if t.length is None else f"varchar({t.length})",
|
||
DateTime: lambda t: "timestamptz" if t.timezone else "timestamp",
|
||
}
|
||
|
||
|
||
def render_orm_type(orm_type: object) -> str:
|
||
"""SQL-тип ORM-колонки в нотации DBML (`numeric(18,2)`, `varchar`, `timestamptz`, имя enum, ...).
|
||
|
||
Классы сравниваются точно: `SmallInteger` — подкласс `Integer`, `Text` — подкласс `String`, `Float` — подкласс
|
||
`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}"
|
||
if isinstance(orm_type, UUID):
|
||
return "uuid" if orm_type.as_uuid else "uuid(as_uuid=False)"
|
||
cls = type(orm_type)
|
||
if cls in _PARAMETRIZED_TYPES:
|
||
return _PARAMETRIZED_TYPES[cls](orm_type)
|
||
# неизвестный класс (в т.ч. `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:
|
||
"""Тип DBML в канонической нотации; `None` — тип не поддержан сверкой."""
|
||
if isinstance(dbml_type, DbmlEnum):
|
||
return str(dbml_type.name)
|
||
name = str(dbml_type).strip().lower()
|
||
numeric = _NUMERIC_RE.fullmatch(name)
|
||
if numeric:
|
||
return f"numeric({numeric.group(1)},{numeric.group(2)})"
|
||
return name if name in SUPPORTED_DBML_TYPES else None
|
||
|
||
|
||
def type_pair(orm_type: object, dbml_type: object) -> tuple[str, str] | None:
|
||
"""(ожидаемый, фактический) тип в нотации DBML; совпадение — равенство строк. `None` — тип DBML не поддержан.
|
||
|
||
Для enum с верным именем и схемой сравниваются ещё и значения в порядке объявления: при расхождении к имени
|
||
приписываются списки значений, чтобы строка отчёта показывала разницу.
|
||
"""
|
||
expected = render_dbml_type(dbml_type)
|
||
if expected is None:
|
||
return None
|
||
actual = render_orm_type(orm_type)
|
||
if isinstance(dbml_type, DbmlEnum) and isinstance(orm_type, SAEnum) and actual == expected:
|
||
expected_values = [item.name for item in dbml_type.items]
|
||
if list(orm_type.enums) != expected_values:
|
||
return f"{expected}{expected_values}", f"{actual}{list(orm_type.enums)}"
|
||
return expected, actual
|
||
|
||
|
||
def type_problem(orm_type: object, dbml_type: object) -> str | None:
|
||
"""Расхождение только типа (`ожидалось X / фактически Y`), сообщение о неподдержанном типе DBML или `None`."""
|
||
pair = type_pair(orm_type, dbml_type)
|
||
if pair is None:
|
||
return f"тип DBML `{dbml_type}` не поддержан сверкой (добавьте в dbml_parity.SUPPORTED_DBML_TYPES)"
|
||
expected, actual = pair
|
||
return None if expected == actual else f"ожидалось {expected} / фактически {actual}"
|
||
|
||
|
||
# --- default ---------------------------------------------------------------------------------------------------
|
||
|
||
|
||
def _canon_expr(text: str) -> str:
|
||
text = " ".join(text.split())
|
||
if text.startswith("'"):
|
||
return text # строковый литерал — регистр значим
|
||
text = text.lower()
|
||
return _SYNONYMS.get(text, text)
|
||
|
||
|
||
def canonical_default_dbml(default: object) -> str:
|
||
"""Канонический вид DBML-default: bool/число/строка-литерал/`Expression`."""
|
||
if isinstance(default, Expression):
|
||
return _canon_expr(str(default.text))
|
||
if isinstance(default, bool):
|
||
return "true" if default else "false"
|
||
if isinstance(default, int | float):
|
||
return str(default)
|
||
return "'" + str(default).replace("'", "''") + "'"
|
||
|
||
|
||
def canonical_default_orm(server_default: object) -> str:
|
||
"""Канонический вид ORM `server_default` (`text(...)`, `func.now()`, строка)."""
|
||
arg = getattr(server_default, "arg", server_default)
|
||
if isinstance(arg, TextClause):
|
||
return _canon_expr(arg.text)
|
||
if isinstance(arg, FunctionElement):
|
||
return _canon_expr(f"{arg.name}()")
|
||
return _canon_expr(str(arg))
|
||
|
||
|
||
# --- колонки ---------------------------------------------------------------------------------------------------
|
||
|
||
|
||
def _nullability(nullable: bool) -> str:
|
||
return "NULL" if nullable else "NOT NULL"
|
||
|
||
|
||
def _yes_no(flag: bool) -> str:
|
||
return "да" if flag else "нет"
|
||
|
||
|
||
def _column_problems(orm_table: Table, dbml_table: DbmlTable) -> list[str]:
|
||
"""Состав, порядок и свойства колонок одной таблицы."""
|
||
name = dbml_table.name
|
||
dbml_names = [c.name for c in dbml_table.columns]
|
||
orm_names = [c.name for c in orm_table.columns]
|
||
problems = [missing(f"{name}.{col}", ORM) for col in dbml_names if col not in orm_names]
|
||
problems += [missing(f"{name}.{col}", DBML) for col in orm_names if col not in dbml_names]
|
||
# порядок сравнивается только по общим колонкам: пропавшая/лишняя колонка уже названа выше
|
||
common_dbml = [col for col in dbml_names if col in orm_names]
|
||
common_orm = [col for col in orm_names if col in dbml_names]
|
||
if common_dbml != common_orm:
|
||
problems.append(mismatch(f"{name}: порядок колонок", common_dbml, common_orm))
|
||
composite_pk = {c.name for i in dbml_table.indexes if i.pk for c in i.subjects}
|
||
for dbml_col in dbml_table.columns:
|
||
if dbml_col.name in orm_table.columns:
|
||
label = f"{name}.{dbml_col.name}"
|
||
problems += _one_column_problems(label, orm_table.columns[dbml_col.name], dbml_col, composite_pk)
|
||
return problems
|
||
|
||
|
||
def _one_column_problems(label: str, orm_col: Column[Any], dbml_col: DbmlColumn, composite_pk: set[str]) -> list[str]:
|
||
problems: list[str] = []
|
||
is_pk = bool(dbml_col.pk) or dbml_col.name in composite_pk
|
||
# тип и nullability — одной строкой, как в спеке: «ожидалось numeric(18,2) NULL / фактически numeric(18,4) NOT NULL»
|
||
pair = type_pair(orm_col.type, dbml_col.type)
|
||
if pair is None:
|
||
problems.append(f"{label}: тип DBML `{dbml_col.type}` не поддержан сверкой")
|
||
else:
|
||
expected_null = _nullability(not (dbml_col.not_null or is_pk))
|
||
actual_null = _nullability(bool(orm_col.nullable))
|
||
expected, actual = f"{pair[0]} {expected_null}", f"{pair[1]} {actual_null}"
|
||
if expected != actual:
|
||
problems.append(mismatch(label, expected, actual))
|
||
if orm_col.primary_key != is_pk:
|
||
problems.append(mismatch(f"{label}: первичный ключ", _yes_no(is_pk), _yes_no(orm_col.primary_key)))
|
||
if bool(orm_col.unique) != bool(dbml_col.unique):
|
||
problems.append(mismatch(f"{label}: unique", _yes_no(bool(dbml_col.unique)), _yes_no(bool(orm_col.unique))))
|
||
expected_default = ABSENT if dbml_col.default is None else canonical_default_dbml(dbml_col.default)
|
||
actual_default = ABSENT if orm_col.server_default is None else canonical_default_orm(orm_col.server_default)
|
||
if expected_default != actual_default:
|
||
problems.append(mismatch(f"{label}: default", expected_default, actual_default))
|
||
return problems
|
||
|
||
|
||
# --- 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 по колонке-ссылке: множество целей `родитель.колонка`, затем у совпавших целей — ON DELETE/ON UPDATE и
|
||
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):
|
||
targets = expected.setdefault((child.name, child_col.name), {})
|
||
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:
|
||
# DDL строится по ForeignKeyConstraint; ForeignKey(ondelete=...) копирует туда же — читаем оба
|
||
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, {}), actual.get(key, {})
|
||
label = f"{key[0]}.{key[1]}: FK"
|
||
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
|
||
|
||
|
||
def _targets(targets: set[str]) -> str:
|
||
return ", ".join(sorted(targets)) or ABSENT
|
||
|
||
|
||
# --- индексы ---------------------------------------------------------------------------------------------------
|
||
|
||
|
||
def _norm_where(where: object) -> str | None:
|
||
if where is None:
|
||
return None
|
||
text = where.text if isinstance(where, TextClause) else str(where)
|
||
return " ".join(text.split()).lower()
|
||
|
||
|
||
def _index_props(unique: bool, where: str | None, nulls_not_distinct: bool) -> str:
|
||
"""Свойства индекса текстом: `unique WHERE is_current NULLS NOT DISTINCT` / `не unique`."""
|
||
parts = ["unique" if unique else "не unique"]
|
||
if where:
|
||
parts.append(f"WHERE {where}")
|
||
if nulls_not_distinct:
|
||
parts.append("NULLS NOT DISTINCT")
|
||
return " ".join(parts)
|
||
|
||
|
||
def _index_problems(orm_table: Table, dbml_table: DbmlTable) -> list[str]:
|
||
"""Индексы, сгруппированные по набору колонок: состав и свойства (unique, WHERE, NULLS NOT DISTINCT)."""
|
||
expected: dict[tuple[str, ...], list[str]] = {}
|
||
for index in dbml_table.indexes:
|
||
if index.pk:
|
||
continue # составной [pk] — первичный ключ, сверяется в колонках
|
||
note = str(index.note) if index.note else ""
|
||
partial = _PARTIAL_RE.search(note)
|
||
props = _index_props(
|
||
bool(index.unique) or partial is not None,
|
||
_norm_where(partial.group(1)) if partial else None,
|
||
# note с `NULLS NOT DISTINCT` => уникальный индекс обязан иметь postgresql_nulls_not_distinct=True
|
||
_NULLS_NOT_DISTINCT_RE.search(note) is not None,
|
||
)
|
||
expected.setdefault(tuple(c.name for c in index.subjects), []).append(props)
|
||
actual: dict[tuple[str, ...], list[str]] = {}
|
||
for orm_index in orm_table.indexes:
|
||
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)})"
|
||
want, have = sorted(expected.get(columns, [])), sorted(actual.get(columns, []))
|
||
if not have:
|
||
problems.append(missing(label, ORM))
|
||
elif not want:
|
||
problems.append(missing(label, DBML))
|
||
elif want != have:
|
||
problems.append(mismatch(label, "; ".join(want), "; ".join(have)))
|
||
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)
|
||
]
|
||
|
||
|
||
# --- точки входа -----------------------------------------------------------------------------------------------
|
||
|
||
|
||
def find_problems(metadata: MetaData, tables: Collection[str], dbml: PyDBML) -> list[str]:
|
||
"""Расхождения ORM (`metadata`) с DBML для таблиц `tables`; пустой список — совпадение."""
|
||
dbml_tables = {t.name: t for t in dbml.tables}
|
||
orm_tables = {t.name: t for t in metadata.tables.values()}
|
||
problems: list[str] = []
|
||
for name in sorted(tables):
|
||
if name not in dbml_tables:
|
||
problems.append(missing(name, DBML))
|
||
elif name not in orm_tables:
|
||
problems.append(missing(name, ORM))
|
||
else:
|
||
if orm_tables[name].schema != SCHEMA:
|
||
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})
|
||
|
||
|
||
def enum_problems(enums: Mapping[str, type[enum.Enum]], dbml: PyDBML) -> list[str]:
|
||
"""Enum'ы ORM-реестра (имя DBML → класс) против блоков `Enum {}` DBML: множество имён и значения по порядку."""
|
||
dbml_enums = {e.name: [item.name for item in e.items] for e in dbml.enums}
|
||
problems: list[str] = []
|
||
for name in sorted(dbml_enums.keys() | enums.keys()):
|
||
if name not in enums:
|
||
problems.append(missing(f"enum {name}", ORM))
|
||
elif name not in dbml_enums:
|
||
problems.append(missing(f"enum {name}", DBML))
|
||
else:
|
||
actual = [member.value for member in enums[name]]
|
||
if actual != dbml_enums[name]:
|
||
problems.append(mismatch(f"enum {name}", dbml_enums[name], actual))
|
||
return problems
|
||
|
||
|
||
def find_all_problems(metadata: MetaData, enums: Mapping[str, type[enum.Enum]], dbml: PyDBML) -> list[str]:
|
||
"""Полная сверка: все таблицы обеих сторон (количество — следствие сравнения множеств) и все enum'ы."""
|
||
names = {t.name for t in dbml.tables} | {t.name for t in metadata.tables.values()}
|
||
return find_problems(metadata, names, dbml) + enum_problems(enums, dbml)
|