feat: add drawdb module

dbterd/adapters/targets/drawdb.py

@@ -0,0 +1,130 @@
+import json
+from typing import Tuple
+
+from dbterd.adapters import adapter
+from dbterd.types import Catalog, Manifest
+
+
+def run(manifest: Manifest, catalog: Catalog, **kwargs) -> Tuple[str, str]:
+    """Parse dbt artifacts and export DDB file
+
+    Args:
+        manifest (dict): Manifest json
+        catalog (dict): Catalog json
+
+    Returns:
+        Tuple(str, str): File name and the DDB (json) content
+    """
+    output_file_name = kwargs.get("output_file_name") or "output.ddb"
+    return (output_file_name, parse(manifest, catalog, **kwargs))
+
+
+def parse(manifest: Manifest, catalog: Catalog, **kwargs) -> str:
+    """Get the DDB content from dbt artifacts
+
+    Args:
+        manifest (dict): Manifest json
+        catalog (dict): Catalog json
+
+    Returns:
+        str: DDB (json) content
+    """
+
+    algo_module = adapter.load_algo(name=kwargs["algo"])
+    tables, relationships = algo_module.parse(
+        manifest=manifest, catalog=catalog, **kwargs
+    )
+
+    # Build DDB content
+    graphic_tables = dict()
+    for idx, x in enumerate(tables):
+        idx_fields = dict()
+        graphic_tables[x.name] = dict(
+            id=idx,
+            x=0,  # TODO
+            y=0,  # TODO
+            fields=idx_fields,
+        )
+        for idc, c in enumerate(x.columns):
+            idx_fields[c.name] = dict(id=idc)
+
+    drawdb = dict(
+        author="Generated by dbterd",
+        title=f"Project ID: {manifest.metadata.project_id}",
+        date=str(manifest.metadata.generated_at),
+        tables=[
+            dict(
+                id=idx,
+                name=x.name,
+                x=graphic_tables.get(x.name, {}).get("x"),
+                y=graphic_tables.get(x.name, {}).get("y"),
+                comment=x.description,
+                indices=[],
+                color="#6360f7",
+                fields=[
+                    dict(
+                        id=idc,
+                        name=c.name,
+                        type=c.data_type,
+                        default="",
+                        check="",
+                        primary=False,  # TODO
+                        unique=False,  # TODO
+                        notNull=False,  # TODO
+                        increment=False,
+                        comment=c.description,
+                    )
+                    for idc, c in enumerate(x.columns)
+                ],
+            )
+            for idx, x in enumerate(tables)
+        ],
+        relationships=[
+            dict(
+                id=idx,
+                name="_".join(x.table_map),
+                cardinality=get_rel_symbol(x.type),
+                startTableId=graphic_tables.get(x.table_map[1], {}).get("id"),
+                endTableId=graphic_tables.get(x.table_map[0], {}).get("id"),
+                startFieldId=(
+                    graphic_tables.get(x.table_map[1], {})
+                    .get("fields")
+                    .get(x.column_map[1], {})
+                    .get("id")
+                ),
+                endFieldId=(
+                    graphic_tables.get(x.table_map[0], {})
+                    .get("fields")
+                    .get(x.column_map[0], {})
+                    .get("id")
+                ),
+                updateConstraint="No action",
+                deleteConstraint="No action",
+            )
+            for idx, x in enumerate(relationships)
+        ],
+        notes=[],
+        subjectAreas=[],
+        database="generic",
+        types=[],
+    )
+
+    return json.dumps(drawdb)
+
+
+def get_rel_symbol(relationship_type: str) -> str:
+    """Get DDB relationship symbol
+
+    Args:
+        relationship_type (str): relationship type
+
+    Returns:
+        str: Relation symbol supported in DDB
+    """
+    if relationship_type in ["01", "11"]:
+        return "One to one"
+    if relationship_type in ["0n", "1n"]:
+        return "One to many"
+    if relationship_type in ["nn"]:
+        return "Many to many"
+    return "Many to one"  # n1