dashsynthetic

 29@dataclass
 30class OntologyConfig:
 31    catalog: str
 32    pk_fk: list[PkFkRelationship] = field(default_factory=list)
 33    reference: list[ReferenceRelationship] = field(default_factory=list)
 34
 35    # ── Mutators ──────────────────────────────────────────────────────────────
 36
 37    def add_pk_fk(
 38        self,
 39        from_table: str,
 40        from_columns: list[str],
 41        to_table: str,
 42        to_columns: list[str],
 43    ) -> "OntologyConfig":
 44        if len(from_columns) != len(to_columns):
 45            raise ValueError(
 46                f"from_columns ({from_columns}) and to_columns ({to_columns}) "
 47                "must have the same length"
 48            )
 49        self.pk_fk.append(
 50            PkFkRelationship(from_table, list(from_columns), to_table, list(to_columns))
 51        )
 52        return self
 53
 54    def add_reference(
 55        self,
 56        from_table: str,
 57        from_column: str,
 58        to_table: str,
 59        to_column: str,
 60    ) -> "OntologyConfig":
 61        self.reference.append(
 62            ReferenceRelationship(from_table, from_column, to_table, to_column)
 63        )
 64        return self
 65
 66    def remove_pk_fk(self, from_table: str, to_table: str) -> int:
 67        """Remove all PK-FK relationships between from_table and to_table. Returns count removed."""
 68        before = len(self.pk_fk)
 69        self.pk_fk = [
 70            r for r in self.pk_fk
 71            if not (r.from_table == from_table and r.to_table == to_table)
 72        ]
 73        return before - len(self.pk_fk)
 74
 75    def remove_reference(self, from_table: str, from_column: str) -> int:
 76        before = len(self.reference)
 77        self.reference = [
 78            r for r in self.reference
 79            if not (r.from_table == from_table and r.from_column == from_column)
 80        ]
 81        return before - len(self.reference)
 82
 83    # ── Queries ───────────────────────────────────────────────────────────────
 84
 85    def parents_of(self, full_table_name: str) -> list[str]:
 86        """All tables that must be generated before this one (deduplicated, ordered)."""
 87        seen: dict[str, None] = {}
 88        for rel in self.pk_fk:
 89            if rel.from_table == full_table_name:
 90                seen[rel.to_table] = None
 91        for rel in self.reference:
 92            if rel.from_table == full_table_name:
 93                seen[rel.to_table] = None
 94        return list(seen)
 95
 96    def fk_relations_for(self, full_table_name: str) -> list[PkFkRelationship]:
 97        return [r for r in self.pk_fk if r.from_table == full_table_name]
 98
 99    def reference_relations_for(self, full_table_name: str) -> list[ReferenceRelationship]:
100        return [r for r in self.reference if r.from_table == full_table_name]
101
102    # ── Topological sort ──────────────────────────────────────────────────────
103
104    def generation_order(self, tables: list[str]) -> list[str]:
105        """
106        Return tables sorted so every table appears after all its parents.
107        Tables referenced in relationships but absent from `tables` are
108        prepended automatically.
109        Raises ValueError on a circular dependency.
110        """
111        all_tables = list(tables)
112        # pull in any dependency not explicitly listed
113        for t in tables:
114            for p in self.parents_of(t):
115                if p not in all_tables:
116                    all_tables.append(p)
117
118        deps: dict[str, list[str]] = {t: self.parents_of(t) for t in all_tables}
119
120        ordered: list[str] = []
121        visited: set[str] = set()
122        visiting: set[str] = set()
123
124        def visit(t: str) -> None:
125            if t in visited:
126                return
127            if t in visiting:
128                raise ValueError(f"Circular dependency detected involving table {t!r}")
129            visiting.add(t)
130            for dep in deps.get(t, []):
131                visit(dep)
132            visiting.discard(t)
133            visited.add(t)
134            ordered.append(t)
135
136        for t in all_tables:
137            visit(t)
138        return ordered
139
140    # ── Serialisation ─────────────────────────────────────────────────────────
141
142    def to_dict(self) -> dict:
143        return {
144            "catalog": self.catalog,
145            "pk_fk": [
146                {
147                    "from_table": r.from_table,
148                    "from_columns": r.from_columns,
149                    "to_table": r.to_table,
150                    "to_columns": r.to_columns,
151                }
152                for r in self.pk_fk
153            ],
154            "reference": [
155                {
156                    "from_table": r.from_table,
157                    "from_column": r.from_column,
158                    "to_table": r.to_table,
159                    "to_column": r.to_column,
160                }
161                for r in self.reference
162            ],
163        }
164
165    @classmethod
166    def from_dict(cls, d: dict) -> "OntologyConfig":
167        config = cls(catalog=d.get("catalog", ""))
168        for r in d.get("pk_fk", []):
169            config.pk_fk.append(
170                PkFkRelationship(
171                    from_table=r["from_table"],
172                    from_columns=r["from_columns"],
173                    to_table=r["to_table"],
174                    to_columns=r["to_columns"],
175                )
176            )
177        for r in d.get("reference", []):
178            config.reference.append(
179                ReferenceRelationship(
180                    from_table=r["from_table"],
181                    from_column=r["from_column"],
182                    to_table=r["to_table"],
183                    to_column=r["to_column"],
184                )
185            )
186        return config
187
188    def validate(self) -> list[str]:
189        issues = []
190        for r in self.pk_fk:
191            if len(r.from_columns) != len(r.to_columns):
192                issues.append(
193                    f"PK-FK {r.from_table}→{r.to_table}: "
194                    f"column counts differ ({r.from_columns} vs {r.to_columns})"
195                )
196        try:
197            all_tables = list({r.from_table for r in self.pk_fk} |
198                               {r.to_table for r in self.pk_fk} |
199                               {r.from_table for r in self.reference} |
200                               {r.to_table for r in self.reference})
201            self.generation_order(all_tables)
202        except ValueError as e:
203            issues.append(str(e))
204        return issues

 16class ProfileStore:
 17    """
 18    Read/write/list table profiles and ontology config stored under base_path.
 19
 20    All methods are pure Python — no Spark, no heavy imports.
 21    """
 22
 23    def __init__(self, base_path: str = "/tmp/dashsynthetic"):
 24        self.base_path = Path(base_path)
 25
 26    # ── Paths ─────────────────────────────────────────────────────────────────
 27
 28    def table_path(self, catalog: str, schema: str, table: str) -> Path:
 29        return self.base_path / catalog / schema / f"{table}.json"
 30
 31    def ontology_path(self, catalog: str) -> Path:
 32        return self.base_path / catalog / "ontology.json"
 33
 34    def catalog_dir(self, catalog: str) -> Path:
 35        return self.base_path / catalog
 36
 37    # ── Profile CRUD ──────────────────────────────────────────────────────────
 38
 39    def save_profile(self, profile: dict) -> Path:
 40        """Write profile dict to disk; creates directories as needed."""
 41        path = self.table_path(profile["catalog"], profile["schema"], profile["table"])
 42        path.parent.mkdir(parents=True, exist_ok=True)
 43        path.write_text(json.dumps(profile, indent=2, default=str))
 44        return path
 45
 46    def load_profile(self, catalog: str, schema: str, table: str) -> Optional[dict]:
 47        """Return the profile dict, or None if it doesn't exist."""
 48        path = self.table_path(catalog, schema, table)
 49        if not path.exists():
 50            return None
 51        return json.loads(path.read_text())
 52
 53    def profile_exists(self, catalog: str, schema: str, table: str) -> bool:
 54        return self.table_path(catalog, schema, table).exists()
 55
 56    def update_last_checked(self, catalog: str, schema: str, table: str, timestamp: str) -> bool:
 57        """Patch only the last_checked field without touching the rest of the profile."""
 58        path = self.table_path(catalog, schema, table)
 59        if not path.exists():
 60            return False
 61        profile = json.loads(path.read_text())
 62        profile["last_checked"] = timestamp
 63        path.write_text(json.dumps(profile, indent=2, default=str))
 64        return True
 65
 66    def list_profiles(
 67        self,
 68        catalog: str,
 69        schema: Optional[str] = None,
 70    ) -> list[dict]:
 71        """
 72        Return all saved profiles for a catalog, optionally filtered by schema.
 73        Skips ontology.json and any unreadable files silently.
 74        """
 75        catalog_dir = self.catalog_dir(catalog)
 76        if not catalog_dir.exists():
 77            return []
 78        profiles = []
 79        for json_path in sorted(catalog_dir.rglob("*.json")):
 80            if json_path.name == "ontology.json":
 81                continue
 82            if schema and json_path.parent.name != schema:
 83                continue
 84            try:
 85                profiles.append(json.loads(json_path.read_text()))
 86            except Exception:
 87                pass
 88        return profiles
 89
 90    def list_catalogs(self) -> list[str]:
 91        if not self.base_path.exists():
 92            return []
 93        return [d.name for d in sorted(self.base_path.iterdir()) if d.is_dir()]
 94
 95    def list_schemas(self, catalog: str) -> list[str]:
 96        catalog_dir = self.catalog_dir(catalog)
 97        if not catalog_dir.exists():
 98            return []
 99        return [
100            d.name for d in sorted(catalog_dir.iterdir())
101            if d.is_dir()
102        ]
103
104    # ── Ontology CRUD ─────────────────────────────────────────────────────────
105
106    def save_ontology(self, catalog: str, ontology: dict) -> Path:
107        path = self.ontology_path(catalog)
108        path.parent.mkdir(parents=True, exist_ok=True)
109        path.write_text(json.dumps(ontology, indent=2, default=str))
110        return path
111
112    def load_ontology(self, catalog: str) -> Optional[dict]:
113        path = self.ontology_path(catalog)
114        if not path.exists():
115            return None
116        return json.loads(path.read_text())