def collect_schema_model()

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

• 16 lines of code
• 3 McCabe index (conditional complexity)

    def collect_schema_model(self, connection, catalog: str, schema: str) -> list[DatabaseTable] | None:
        comps = self._component_queries()
        results: dict[str, list[dict]] = {cq: [] for cq in comps}

        for cq, template_sql in comps.items():
            sql = template_sql.replace("{SCHEMA}", self._quote_literal(schema))
            rows = self._fetchall_dicts(connection, sql, None)
            results[cq] = rows

        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", []),
        )