# How to Connect Snowflake to Semantica for Knowledge Graph Ingestion

> Easily connect Snowflake to Semantica for knowledge graph ingestion using the native SnowflakeIngestor class. Supports password, key-pair, OAuth, and SSO authentication for seamless data import.

- Repository: [Semantica /semantica](https://github.com/semantica-agi/semantica)
- Tags: how-to-guide
- Published: 2026-08-15

---

**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**:

```bash
pip install "semantica[db-snowflake]"

```

Or install the connector standalone:

```bash
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)](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.

```python
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).

```python
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.

```python
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.

```python
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

```python
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

```python
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:

```python
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.

```python
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:

```python
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`:

```python
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`](https://github.com/semantica-agi/semantica/blob/main/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`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/snowflake_ingestor.py) with comprehensive tests in [`tests/test_snowflake_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/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`](https://github.com/semantica-agi/semantica/blob/main/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`](https://github.com/semantica-agi/semantica/blob/main/tests/test_snowflake_ingestor.py) demonstrate mocked connection flows and edge‑case handling without requiring a live Snowflake warehouse.