def collect_schema_model()

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

• 40 lines of code
• 4 McCabe index (conditional complexity)

    def collect_schema_model(self, connection, catalog: str, schema: str) -> list[DatabaseTable] | None:
        q = self._sql_columns(catalog, schema)
        rows = self._fetchall_dicts(connection, q.sql, q.params)
        if not rows:
            return []

        rels: list[dict] = []
        cols: list[dict] = []
        seen_tables: set[str] = set()

        for r in rows:
            table_name = r["table_name"]
            if table_name not in seen_tables:
                seen_tables.add(table_name)
                rels.append(
                    {
                        "table_name": table_name,
                        "kind": "table",
                        "description": None,
                    }
                )

            cols.append(
                {
                    "table_name": table_name,
                    "column_name": r["column_name"],
                    "ordinal_position": r["ordinal_position"],
                    "data_type": r["data_type"],
                    "is_nullable": True,
                    "default_expression": None,
                    "generated": None,
                    "description": None,
                }
            )

        return TableBuilder.build_from_components(
            rels=rels,
            cols=cols,
            pk_cols=[],
            uq_cols=[],
            checks=[],
            fk_cols=[],
            idx_cols=[],
        )