def collect_schema_model()

in src/databao_context_engine/plugins/databases/mssql_introspector.py [0:0]

• 31 lines of code
• 11 McCabe index (conditional complexity)

    def collect_schema_model(self, connection, catalog: str, schema: str) -> list[DatabaseTable] | None:
        comps = self._component_queries()

        schema_lit = self._quote_literal(schema)
        stmts = [sql.replace("{SCHEMA}", schema_lit) for sql in comps.values()]
        batch_prefix = "SET NOCOUNT ON; SET XACT_ABORT ON;"
        batch = batch_prefix + "\n" + ";\n".join(s.rstrip().rstrip(";") for s in stmts) + ";"

        results: dict[str, list[dict]] = {name: [] for name in comps}
        with connection.cursor() as cur:
            cur.execute(batch)
            for ix, name in enumerate(comps.keys(), start=1):
                try:
                    rows: list[dict] = []
                    if cur.description:
                        cols = [c[0].lower() for c in cur.description]
                        rows = [dict(zip(cols, r)) for r in cur.fetchall()]
                    results[name] = rows
                except Exception as e:
                    raise RuntimeError(f"Failed reading component #{ix} '{name}'") from e
                if ix < len(comps):
                    ok = cur.nextset()
                    if not ok:
                        raise RuntimeError(f"Batch ended early after component #{ix} '{name}'")

        return TableBuilder.build_from_components(
            rels=results.get("relations", []),
            cols=results.get("columns", []),
            pk_cols=results.get("pk", []),
            uq_cols=results.get("uq", []),
            checks=results.get("checks", []),
            fk_cols=results.get("fks", []),
            idx_cols=results.get("idx", []),
        )