Skip to content

ingester

module wittgenstein_schema_discovery.ingester

CatalogIngester — indexes DDL + documentation from catalog.yaml / domain_glossary.yaml into pgvector.

Writes to the 'ddl' and 'documentation' LangChain PGVector collections. Uses psycopg2 for the clear_collections() operation (direct SQL delete).

Classes

  • CatalogIngester — Ingests catalog YAML and glossary YAML into the pgvector 'ddl' and 'documentation' collections.

wittgenstein_schema_discovery.ingester.CatalogIngester

class CatalogIngester(connection_string: str, pg_dsn: str, embedder: Embeddings)

Ingests catalog YAML and glossary YAML into the pgvector 'ddl' and 'documentation' collections.

Each call to ingest_catalog() or ingest_glossary() appends to the existing collections. Call clear_collections() before re-ingesting to avoid duplicate embeddings.

Parameters

  • connection_string : str — psycopg3 URL for LangChain PGVector (e.g. 'postgresql+psycopg://user:pass@host/db').

  • pg_dsn : str — psycopg2 DSN for the clear_collections direct-SQL operation (e.g. 'postgresql://user:pass@host:port/db').

  • embedder : Embeddings — any LangChain Embeddings instance.

Methods

  • ingest_catalog — Index DDL + table/column/FK docs from catalog.yaml.

  • ingest_glossary — Index all glossary entries (phases, typologies, terms) from domain_glossary.yaml.

  • clear_collections — Delete all embeddings from the 'sql', 'ddl', and 'documentation' collections.

wittgenstein_schema_discovery.ingester.CatalogIngester.ingest_catalog

method CatalogIngester.ingest_catalog(yaml_path: Path, db_tables: set[str]) → dict[str, int]

Index DDL + table/column/FK docs from catalog.yaml.

Only tables that appear in db_tables are indexed (skips tables absent from the actual DB to avoid confusing the agent).

Parameters

  • yaml_path : Path — path to catalog.yaml.

  • db_tables : set[str] — set of table names that exist in the target DB (from wittgenstein_sql_runner.discover_db_tables()).

Returns

  • Dict with counts — {'ddl', 'doc_table', 'doc_col', 'fk', 'skipped'}.

wittgenstein_schema_discovery.ingester.CatalogIngester.ingest_glossary

method CatalogIngester.ingest_glossary(yaml_path: Path) → int

Index all glossary entries (phases, typologies, terms) from domain_glossary.yaml.

Parameters

  • yaml_path : Path — path to domain_glossary.yaml.

Returns

  • int — Number of entries indexed.

wittgenstein_schema_discovery.ingester.CatalogIngester.clear_collections

method CatalogIngester.clear_collections() → int

Delete all embeddings from the 'sql', 'ddl', and 'documentation' collections.

Returns the count of removed rows. Returns 0 if LangChain tables don't exist yet.