How to Connect Snowflake to Semantica for Knowledge Graph Ingestion

Semantica provides a native SnowflakeIngestor class that pulls warehouse data, converts rows into documents, and feeds them directly into the knowledge graph pipeline with support for password, key‑pair, OAuth, and SSO authentication.

The semantica-agi/semantica repository ships a production‑ready Snowflake integration under the semantica.ingest package. This guide walks through authenticating, extracting tables or custom queries, and converting results into the JSON‑compatible documents required by Semantica's KG builder.

Installation

Install Semantica with Snowflake support using the extra dependency:

pip install "semantica[db-snowflake]"

Or install the connector standalone:

pip install snowflake-connector-python

See the installation notes in [docs/integrations/snowflake.md](https://github.com/semantica-agi/semantica/blob/main/docs/integrations/snowflake.md#L10‑L18).

Authentication Options for Snowflake Connection

Semantica supports four authentication methods, all configurable via constructor arguments or environment variables (SNOWFLAKE_ACCOUNT, SNOWFLAKE_USER, etc.). Environment variables are recommended to keep secrets out of source code.

Password Authentication

Provide account, user, and password directly or via env vars.

from semantica.ingest import SnowflakeIngestor

ingestor = SnowflakeIngestor(
    account="myaccount",
    user="myuser",
    password="mypassword",
    warehouse="COMPUTE_WH",
    database="MYDB",
)

Relevant source: SnowflakeConnector.__init__ (lines 99‑108) and password handling in connect.

Key‑Pair Authentication (Recommended)

Supply private_key_path to a PEM file. The connector loads and decrypts the key via _load_private_key (lines 84‑102).

ingestor = SnowflakeIngestor(
    account="myaccount",
    user="myuser",
    private_key_path="/path/to/rsa_key.p8",
    warehouse="COMPUTE_WH",
)

OAuth Authentication

Set authenticator="oauth" with a valid token.

ingestor = SnowflakeIngestor(
    account="myaccount",
    user="myuser",
    authenticator="oauth",
    token="my_oauth_token",
)

OAuth handling appears in connect (lines 158‑166).

SSO (External Browser)

Use authenticator="externalbrowser" to trigger browser‑based login.

ingestor = SnowflakeIngestor(
    account="myaccount",
    user="myuser",
    authenticator="externalbrowser",
)

if ingestor.connector.test_connection():
    print("SSO connection established")

Ingesting Snowflake Data into Semantica

The SnowflakeIngestor class provides two main extraction paths: full‑table ingestion with optional filtering, and arbitrary SQL queries. Both return a SnowflakeData dataclass (lines 61‑73) that stores raw rows, column metadata, and ingestion timestamps.

Ingest a Table with Filters

from semantica.ingest import SnowflakeIngestor
import os

ingestor = SnowflakeIngestor(
    account=os.getenv("SNOWFLAKE_ACCOUNT"),
    user=os.getenv("SNOWFLAKE_USER"),
    password=os.getenv("SNOWFLAKE_PASSWORD"),
    warehouse=os.getenv("SNOWFLAKE_WAREHOUSE"),
    database=os.getenv("SNOWFLAKE_DATABASE"),
    schema=os.getenv("SNOWFLAKE_SCHEMA"),
)

customers = ingestor.ingest_table(
    "CUSTOMERS",
    where="COUNTRY = 'USA' AND CREATED_DATE > '2024-01-01'",
    order_by="CREATED_DATE DESC",
    limit=10_000,
)

print(f"Rows: {customers.row_count}, columns: {customers.columns}")

Implementation: ingest_table (lines 438‑575) with integrated progress tracking via start_tracking / update_tracking / stop_tracking.

Run a Custom Query

sales = ingestor.ingest_query(
    """
    SELECT CUSTOMER_ID, SUM(AMOUNT) AS TOTAL_AMOUNT
    FROM SALES
    WHERE DATE >= '2024-01-01'
    GROUP BY CUSTOMER_ID
    ORDER BY TOTAL_AMOUNT DESC
    """,
    batch_size=5_000,  # optional server‑side batching

)

Implementation: ingest_query (lines 582‑674).

Schema Introspection

Inspect column types before building your KG schema:

schema = ingestor.get_table_schema("CUSTOMERS")
for col in schema["columns"]:
    print(f"{col['name']}: {col['type']} (nullable={col['nullable']})")

Uses INFORMATION_SCHEMA.COLUMNS (lines 718‑730) and fetches primary‑key columns (lines 736‑746).

Converting Snowflake Data to KG Documents

The export_as_documents method (lines 826‑876) transforms SnowflakeData into the list‑of‑dicts format required by Semantica's pipeline.

docs = ingestor.export_as_documents(
    customers,
    id_field="CUSTOMER_ID",
    text_fields=["NAME", "EMAIL", "NOTES"],
    metadata_fields=["COUNTRY", "CREATED_DATE"],
)

print(f"Created {len(docs)} documents ready for KG ingestion")

Feed these documents directly into your pipeline:

from semantica.pipeline import KnowledgeGraphBuilder

kg_builder = KnowledgeGraphBuilder()
kg = kg_builder.build_from_documents(docs)

Batch Processing Large Snowflake Tables

For tables exceeding memory limits, paginate with limit and offset:

PAGE_SIZE = 5_000

for page in range(100):
    page_data = ingestor.ingest_table(
        "BIG_TABLE",
        limit=PAGE_SIZE,
        offset=page * PAGE_SIZE,
    )
    docs = ingestor.export_as_documents(page_data)
    kg_builder.add_batch(docs)

Alternatively, ingest_query supports native batch_size parameter for server‑side result chunking.

Connecting Snowflake to Semantica: Architecture Overview

Component	Role	Key Source Location
SnowflakeConnector	Connection management, authentication routing, test_connection() helper	snowflake_ingestor.py lines 99‑114, 124‑184
SnowflakeIngestor	High‑level façade for table/query ingestion, schema introspection, document export	Constructor lines 60‑66, ingest_table 438‑575, ingest_query 582‑674, export_as_documents 826‑876
SnowflakeData	Dataclass holding rows, columns, metadata	Lines 61‑73
Progress Tracker	Per‑operation status reporting and failure handling	Integrated in ingest_table and ingest_query

The data flow: SnowflakeConnector establishes connection → SnowflakeIngestor executes extraction → SnowflakeData stores results → export_as_documents produces JSON‑compatible docs → KG pipeline consumes documents.

Common Pitfalls When Connecting Snowflake to Semantica

• Missing credentials — SnowflakeConnector raises ValidationError if account or user is absent (lines 156‑168). Always verify environment variables are set.
• SQL injection in filters — The where and order_by parameters are concatenated directly into queries. Semicolons are blocked (lines 108‑113), but only pass trusted literals.
• Manual identifier quoting — Use the built‑in _escape_identifier helper (lines 125‑136) for database, schema, and table names rather than adding quotes yourself.

Summary

• Install with pip install "semantica[db-snowflake]".
• Authenticate via password, key‑pair, OAuth, or SSO; prefer environment variables for secrets.
• Use ingest_table() for filtered table extracts or ingest_query() for custom SQL.
• Convert results to KG‑ready documents with export_as_documents().
• Leverage get_table_schema() for column introspection and limit/offset for large‑table pagination.
• All core logic resides in semantica/ingest/snowflake_ingestor.py with comprehensive tests in tests/test_snowflake_ingestor.py.

Frequently Asked Questions

What authentication method should I use for production Snowflake connections?

Key‑pair authentication is recommended for production. It eliminates password rotation, supports automated CI/CD pipelines, and is implemented via private_key_path or private_key bytes. The connector loads PEM files securely through _load_private_key (lines 84‑102) without exposing credentials in process lists.

How do I handle very large tables without loading everything into memory?

Use the limit and offset parameters of ingest_table() to paginate client‑side, or specify batch_size in ingest_query() for server‑side chunking. Both approaches return SnowflakeData objects that you convert to documents incrementally and feed to KnowledgeGraphBuilder.add_batch() rather than build_from_documents().

Can I trust the where clause parameter in ingest_table()?

No — where strings are concatenated directly into generated SQL. Semantica blocks semicolons (lines 108‑113) as a basic safeguard, but you must sanitize or validate any user‑provided filters before passing them. Treat where and order_by as trusted‑code‑only parameters.

Where can I find working code examples for Snowflake KG ingestion?

The cookbook/advanced/snowflake_ingestion_examples.py file contains runnable patterns for password auth, key‑pair auth, OAuth, SSO, and batch processing. Unit tests in tests/test_snowflake_ingestor.py demonstrate mocked connection flows and edge‑case handling without requiring a live Snowflake warehouse.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}