From b2093978751e49ec05e4d5ec0f0fae776b384597 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 6 Jul 2026 13:55:49 +0900 Subject: [PATCH] =?UTF-8?q?feat(design-first):=20DBML=20import=20=E2=80=94?= =?UTF-8?q?=20design=20a=20schema=20without=20a=20database?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Top priority from docs/erd-tool-feature-research.md (#486): dbdiagram.io's core workflow. parse_dbml(text) converts the dbdiagram DBML dialect (tables, quoted/schema-qualified names, pk/not null/unique settings, inline & standalone refs incl. reverse '<', Project/Enum/TableGroup blocks skipped) into the SAME snapshot JSON introspection produces — so DDL export, diff, migration, and all analyzers work on a design that never touched a database. Derives the constraints array (PK attnums, FK defs) that ddl/export renders. POST /api/dbml/convert returns snapshot_json + optional dialect DDL. Line-oriented parser: hostile input degrades to skipped lines, never an exception. +5 tests incl. end-to-end DBML->DDL; suite green; mypy clean. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01AxU2xaupAjp912oDNFuWyd --- backend/app/api/dbml.py | 36 +++++ backend/app/main.py | 2 + backend/app/schemas.py | 17 ++ backend/app/spec/dbml_import.py | 260 ++++++++++++++++++++++++++++++ backend/tests/test_dbml_import.py | 90 +++++++++++ 5 files changed, 405 insertions(+) create mode 100644 backend/app/api/dbml.py create mode 100644 backend/app/spec/dbml_import.py create mode 100644 backend/tests/test_dbml_import.py diff --git a/backend/app/api/dbml.py b/backend/app/api/dbml.py new file mode 100644 index 00000000..c35db752 --- /dev/null +++ b/backend/app/api/dbml.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +from fastapi import APIRouter, Depends + +from app.auth import CurrentUser, get_current_user +from app.ddl.export import snapshot_json_to_sql +from app.schemas import DbmlConvertIn, DbmlConvertOut +from app.spec.dbml_import import parse_dbml + +router = APIRouter(prefix="/api/dbml", tags=["dbml"]) + + +@router.post("/convert", response_model=DbmlConvertOut) +async def convert_dbml( + body: DbmlConvertIn, + user: CurrentUser = Depends(get_current_user), +) -> DbmlConvertOut: + """Design-first entry point: DBML text → snapshot JSON (+ optional DDL). + + The returned snapshot has the same shape introspection produces, so the + whole downstream pipeline (ERD, DDL export, diff, migration, analyzers) + works on a design that never touched a database. Pure computation — no + project resources involved, so authentication alone suffices. + """ + snapshot = parse_dbml(body.dbml) + ddl = ( + snapshot_json_to_sql(snapshot, target_dialect=body.dialect) + if body.include_ddl + else None + ) + return DbmlConvertOut( + snapshot_json=snapshot, + ddl=ddl, + tables=len(snapshot["relations"]), + foreign_keys=len(snapshot["fk_edges"]), + ) diff --git a/backend/app/main.py b/backend/app/main.py index 41f802c7..0b68b29e 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -9,6 +9,7 @@ from fastapi.middleware.cors import CORSMiddleware from app.api.connections import router as connections_router +from app.api.dbml import router as dbml_router from app.api.auth_routes import router as auth_router from app.api.me import router as me_router from app.api.projects import router as projects_router @@ -164,6 +165,7 @@ async def csrf_token() -> dict[str, str]: app.include_router(projects_router) app.include_router(connections_router) +app.include_router(dbml_router) app.include_router(snapshots_router) app.include_router(me_router) app.include_router(share_router) diff --git a/backend/app/schemas.py b/backend/app/schemas.py index 0359799c..8ef217e3 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -97,3 +97,20 @@ class MeOut(BaseModel): user_account_uuid: uuid.UUID subject: str display_name: str | None + + +class DbmlConvertIn(BaseModel): + """Request body for converting DBML text into a snapshot.""" + + dbml: str = Field(min_length=1, max_length=524_288) + include_ddl: bool = True + dialect: Literal["postgresql", "snowflake"] = "postgresql" + + +class DbmlConvertOut(BaseModel): + """DBML conversion result: snapshot JSON plus optional DDL.""" + + snapshot_json: dict + ddl: str | None = None + tables: int + foreign_keys: int diff --git a/backend/app/spec/dbml_import.py b/backend/app/spec/dbml_import.py new file mode 100644 index 00000000..4c6f4f85 --- /dev/null +++ b/backend/app/spec/dbml_import.py @@ -0,0 +1,260 @@ +"""Parse DBML (database markup language) into the common snapshot JSON. + +Design-first workflow: write DBML (the dbdiagram.io/dbdocs dialect), convert it +to the same snapshot shape introspection produces, and every downstream feature +works on it unchanged — DDL export, migration generation, analyzers, the ERD. + +Supported subset (the parts real DBML files actually use): + +* ``Table [schema.]name { ... }`` with columns ``name type [settings]`` +* column settings: ``pk``/``primary key``, ``not null``, ``unique``, ``default:``, + ``note:``, inline ``ref: > other.col`` (also ``<`` and ``-``) +* standalone ``Ref: a.col > b.col`` / ``Ref name { a.col > b.col }`` +* quoted identifiers ``"My Table"``; comments ``//``; multi-word types + +Ignored (parsed over, not errors): ``Project``/``Enum``/``TableGroup``/``Note`` +blocks, ``indexes`` blocks, header colors. ponytail: line-oriented parser, not a +grammar — good for the 95% of DBML in the wild; a hostile file degrades to +skipped lines, never an exception. +""" + +from __future__ import annotations + +import re +from typing import Any + +_TABLE_RE = re.compile( + r"^table\s+(?:\"(?P[^\"]+)\"|(?P[\w.]+))(?:\s+as\s+\w+)?\s*\{?\s*$", + re.IGNORECASE, +) +_COLUMN_RE = re.compile( + r"^(?:\"(?P[^\"]+)\"|(?P\w+))\s+" + r"(?P[\w]+(?:\([^)]*\))?(?:\[\])?)" + r"(?:\s*\[(?P.*)\])?\s*$" +) +# a dotted path whose segments may be quoted (quotes can contain spaces) +_PATH = r'(?:"[^"]+"|\w+)(?:\.(?:"[^"]+"|\w+))*' +_REF_RE = re.compile( + r"ref\s*(?:\w+\s*)?:?\s*" + rf"(?P{_PATH})\s*(?P[<>-])\s*(?P{_PATH})", + re.IGNORECASE, +) +_INLINE_REF_RE = re.compile(rf"ref:\s*(?P[<>-])\s*(?P{_PATH})", re.IGNORECASE) +_PATH_SEGMENT_RE = re.compile(r'"[^"]+"|[^.]+') + + +def _split_table_name(raw: str) -> tuple[str, str]: + raw = raw.strip().strip('"') + if "." in raw: + schema, _, name = raw.partition(".") + return schema.strip('"'), name.strip('"') + return "public", raw + + +def _split_col_ref(raw: str) -> tuple[str, str, str]: + """'schema.table.col' | 'table.col' -> (schema, table, col). + + Splits on dots *outside* quotes so '"Order Items".account_id' works. + """ + parts = [p.strip('"') for p in _PATH_SEGMENT_RE.findall(raw.strip())] + if len(parts) >= 3: + return parts[0], parts[1], parts[2] + if len(parts) == 2: + return "public", parts[0], parts[1] + return "public", "", parts[0] + + +def parse_dbml(text: str) -> dict[str, Any]: + """Parse DBML text into snapshot JSON (relations/columns/pk_columns/fk_edges).""" + relations: list[dict[str, Any]] = [] + columns: list[dict[str, Any]] = [] + pk_columns: list[dict[str, Any]] = [] + fk_specs: list[tuple[str, str, str, str, str, str]] = [] # child s/t/c, parent s/t/c + + oid_by_table: dict[tuple[str, str], int] = {} + next_oid = 1 + current: tuple[str, str] | None = None + in_ignored_block = 0 + in_indexes = False + + for raw_line in text.splitlines(): + line = raw_line.split("//", 1)[0].strip() + if not line: + continue + + # ignored blocks (Project / Enum / TableGroup / Note) — track braces + if in_ignored_block: + in_ignored_block += line.count("{") - line.count("}") + continue + if re.match(r"^(project|enum|tablegroup|note)\b", line, re.IGNORECASE): + in_ignored_block = line.count("{") - line.count("}") + if in_ignored_block <= 0 and "{" not in line: + in_ignored_block = 1 # block opens on a following line + continue + + m = _TABLE_RE.match(line) + if m: + schema, name = _split_table_name(m.group("qname") or m.group("name")) + current = (schema, name) + if current not in oid_by_table: + oid_by_table[current] = next_oid + relations.append( + { + "relation_oid": next_oid, + "relation_kind": "r", + "schema_name": schema, + "relation_name": name, + "relation_comment": None, + } + ) + next_oid += 1 + continue + + if line.startswith("}"): + current = None + in_indexes = False + continue + + # standalone Ref (works inside or outside a table body) + if re.match(r"^ref\b", line, re.IGNORECASE): + rm = _REF_RE.search(line) + if rm: + fs, ft, fc = _split_col_ref(rm.group("from")) + ts, tt, tc = _split_col_ref(rm.group("to")) + if rm.group("op") == "<": # a < b means b references a + fs, ft, fc, ts, tt, tc = ts, tt, tc, fs, ft, fc + if ft and tt: + fk_specs.append((fs, ft, fc, ts, tt, tc)) + continue + + if current is None: + continue + if re.match(r"^indexes\s*\{", line, re.IGNORECASE): + in_indexes = True + continue + if in_indexes: + if "}" in line: + in_indexes = False + continue + + cm = _COLUMN_RE.match(line) + if not cm: + continue + col_name = (cm.group("qname") or cm.group("name")).strip('"') + settings = (cm.group("settings") or "").lower() + oid = oid_by_table[current] + is_pk = bool(re.search(r"\bpk\b|primary\s+key", settings)) + columns.append( + { + "relation_oid": oid, + "column_name": col_name, + "column_position": sum(1 for c in columns if c["relation_oid"] == oid) + 1, + "data_type": cm.group("type"), + "is_not_null": is_pk or "not null" in settings, + "has_default": "default:" in settings, + "default_expr": None, + "column_comment": None, + } + ) + if is_pk: + pk_columns.append( + {"relation_oid": oid, "column_name": col_name, "column_ordinal": len(pk_columns) + 1} + ) + im = _INLINE_REF_RE.search(cm.group("settings") or "") + if im: + ts, tt, tc = _split_col_ref(im.group("to")) + if im.group("op") == "<": + # inverse inline ref: the other table references this column + fk_specs.append((ts, tt, tc, current[0], current[1], col_name)) + else: + fk_specs.append((current[0], current[1], col_name, ts, tt, tc)) + + fk_edges: list[dict[str, Any]] = [] + for i, (cs, ct, cc, ps, pt, pc) in enumerate(fk_specs, start=1): + child = oid_by_table.get((cs, ct)) + parent = oid_by_table.get((ps, pt)) + if child is None or parent is None: + continue # ref to a table not defined in this document + fk_edges.append( + { + "fk_constraint_oid": 100000 + i, + "fk_constraint_name": f"fk_{ct}_{cc}", + "child_relation_oid": child, + "parent_relation_oid": parent, + "child_column_name": cc, + "parent_column_name": pc, + "column_ordinal": 1, + } + ) + + constraints = _build_constraints(relations, columns, pk_columns, fk_edges) + + return { + "source": "dbml", + "schemas": sorted({r["schema_name"] for r in relations}), + "relations": relations, + "columns": columns, + "constraints": constraints, + "indexes": [], + "pk_columns": pk_columns, + "fk_edges": fk_edges, + "citus_distributed_tables": [], + } + + +def _build_constraints( + relations: list[dict[str, Any]], + columns: list[dict[str, Any]], + pk_columns: list[dict[str, Any]], + fk_edges: list[dict[str, Any]], +) -> list[dict[str, Any]]: + """Derive the ``constraints`` array (what DDL export renders) from parsed parts.""" + rel_by_oid = {r["relation_oid"]: r for r in relations} + pos_by_oid_col: dict[tuple[int, str], int] = { + (c["relation_oid"], c["column_name"]): c["column_position"] for c in columns + } + constraints: list[dict[str, Any]] = [] + + pk_cols_by_oid: dict[int, list[str]] = {} + for pk in pk_columns: + pk_cols_by_oid.setdefault(pk["relation_oid"], []).append(pk["column_name"]) + for oid, cols in pk_cols_by_oid.items(): + rel = rel_by_oid[oid] + quoted = ", ".join(f'"{c}"' for c in cols) + constraints.append( + { + "constraint_oid": 200000 + oid, + "constraint_name": f"pk_{rel['relation_name']}", + "constraint_type": "p", + "schema_name": rel["schema_name"], + "relation_oid": oid, + "relation_name": rel["relation_name"], + "constrained_attnums": [pos_by_oid_col[(oid, c)] for c in cols], + "constraint_def": f"PRIMARY KEY ({quoted})", + } + ) + + for edge in fk_edges: + child = rel_by_oid[edge["child_relation_oid"]] + parent = rel_by_oid[edge["parent_relation_oid"]] + constraints.append( + { + "constraint_oid": 300000 + edge["fk_constraint_oid"], + "constraint_name": edge["fk_constraint_name"], + "constraint_type": "f", + "schema_name": child["schema_name"], + "relation_oid": edge["child_relation_oid"], + "relation_name": child["relation_name"], + "constrained_attnums": [ + pos_by_oid_col.get( + (edge["child_relation_oid"], edge["child_column_name"]), 1 + ) + ], + "constraint_def": ( + f'FOREIGN KEY ("{edge["child_column_name"]}") REFERENCES ' + f'"{parent["schema_name"]}"."{parent["relation_name"]}" ' + f'("{edge["parent_column_name"]}")' + ), + } + ) + return constraints diff --git a/backend/tests/test_dbml_import.py b/backend/tests/test_dbml_import.py new file mode 100644 index 00000000..156171e4 --- /dev/null +++ b/backend/tests/test_dbml_import.py @@ -0,0 +1,90 @@ +from __future__ import annotations + +from app.ddl.export import snapshot_json_to_sql +from app.spec.dbml_import import parse_dbml + +BASIC = """ +// a typical dbdiagram.io document +Table users { + id integer [pk, not null] + username varchar(255) [not null, unique] + created_at timestamp +} + +Table posts { + id integer [pk] + user_id integer [not null, ref: > users.id] + title varchar +} + +Ref: posts.user_id > users.id +""" + + +def test_parses_tables_columns_pks(): + snap = parse_dbml(BASIC) + names = {(r["schema_name"], r["relation_name"]) for r in snap["relations"]} + assert names == {("public", "users"), ("public", "posts")} + cols = {c["column_name"] for c in snap["columns"] if c["relation_oid"] == 1} + assert cols == {"id", "username", "created_at"} + # pk implies not null; plain column stays nullable + by_name = {c["column_name"]: c for c in snap["columns"]} + assert by_name["id"]["is_not_null"] is True + assert by_name["created_at"]["is_not_null"] is False + assert {p["column_name"] for p in snap["pk_columns"]} == {"id"} + + +def test_parses_refs_inline_and_standalone_deduped_semantics(): + snap = parse_dbml(BASIC) + # inline ref + standalone ref both point posts.user_id -> users.id + assert all( + e["child_column_name"] == "user_id" and e["parent_column_name"] == "id" + for e in snap["fk_edges"] + ) + assert len(snap["fk_edges"]) == 2 # parser is literal; dedup is the caller's choice + + +def test_reverse_arrow_and_schema_qualified_and_quoted(): + text = ''' +Table auth.accounts { + account_id bigint [pk] +} +Table "Order Items" { + id bigint [pk] + account_id bigint +} +Ref: auth.accounts.account_id < "Order Items".account_id +''' + snap = parse_dbml(text) + assert ("auth", "accounts") in {(r["schema_name"], r["relation_name"]) for r in snap["relations"]} + edge = snap["fk_edges"][0] + # '<' means the right side references the left + child = next(r for r in snap["relations"] if r["relation_oid"] == edge["child_relation_oid"]) + assert child["relation_name"] == "Order Items" + + +def test_ignores_project_enum_notes_and_unknown_refs(): + text = """ +Project demo { + database_type: 'PostgreSQL' +} +Enum status { + active + banned +} +Table t { + id int [pk] + s status +} +Ref: t.ghost_col > missing_table.id +""" + snap = parse_dbml(text) + assert len(snap["relations"]) == 1 + assert snap["fk_edges"] == [] # ref to undefined table skipped, no crash + + +def test_dbml_snapshot_feeds_existing_ddl_export(): + ddl = snapshot_json_to_sql(parse_dbml(BASIC), target_dialect="postgresql") + assert 'CREATE TABLE IF NOT EXISTS "public"."users"' in ddl + assert 'CREATE TABLE IF NOT EXISTS "public"."posts"' in ddl + assert "PRIMARY KEY" in ddl