# How to Integrate Semantica with Snowflake: A Complete Guide to Data Ingestion

> Learn how to integrate Semantica with Snowflake for seamless data ingestion. This guide details connecting to Snowflake warehouses and converting query results for AI processing.

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

---

**Semantica provides a built-in Snowflake ingestion module that connects to Snowflake warehouses, executes queries, and converts results into Semantica documents for downstream AI processing.**

The `semantica-agi/semantica` repository ships with a native Snowflake integration located in [`semantica/ingest/snowflake_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/snowflake_ingestor.py). This module handles authentication, query execution, and document conversion—enabling you to pull enterprise data directly into semantic pipelines for embedding, reasoning, and retrieval-augmented generation (RAG).

## Prerequisites

Before connecting Semantica to Snowflake, install the official Python connector:

```bash
pip install snowflake-connector-python

```

The Snowflake integration has no additional Semantica dependencies beyond this driver.

## Authentication Methods

The `SnowflakeConnector` class supports multiple authentication strategies as implemented in [`semantica/ingest/snowflake_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/snowflake_ingestor.py). Credentials can be passed explicitly or loaded from environment variables.

### Environment Variable Configuration

Set these variables to avoid hardcoding credentials:

- `SNOWFLAKE_ACCOUNT`
- `SNOWFLAKE_USER`
- `SNOWFLAKE_PASSWORD`
- `SNOWFLAKE_WAREHOUSE`
- `SNOWFLAKE_DATABASE`
- `SNOWFLAKE_SCHEMA`

### Explicit Credential Passing

Instantiate `SnowflakeIngestor` with direct parameters:

```python
from semantica.ingest import SnowflakeIngestor

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

```

### OAuth Authentication

For SSO environments, use OAuth tokens:

```python
ingestor = SnowflakeIngestor(
    account="myaccount",
    user="myuser",
    authenticator="oauth",
    token="my_oauth_access_token",
)

```

### Key-Pair Authentication

For production deployments, key-pair authentication provides enhanced security:

```python
with open("private_key.pem", "rb") as f:
    pk_bytes = f.read()

ingestor = SnowflakeIngestor(
    account="myaccount",
    user="myuser",
    authenticator="snowflake_jwt",
    private_key=pk_bytes,
)

```

The `SnowflakeConnector.__init__` method validates required fields and `SnowflakeConnector.connect` builds the appropriate `snowflake.connector.connect` call based on your authentication method.

## Ingesting Data from Snowflake

### Ingest an Entire Table

The `SnowflakeIngestor.ingest_table` method constructs optimized `SELECT` queries with optional filtering and pagination:

```python
from semantica.ingest import SnowflakeIngestor

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

# Ingest with server-side limits and offsets

data = ingestor.ingest_table(
    "CUSTOMERS",
    where="CREATED_AT > '2024-01-01'",
    order_by="CUSTOMER_ID",
    limit=5000,
    offset=0
)

print(f"Rows retrieved: {data.row_count}")
print(f"Columns: {data.columns}")

```

This method safely constructs `SELECT * FROM <db>.<schema>.<table>` with parameterized `WHERE`, `ORDER BY`, `LIMIT`, and `OFFSET` clauses, then streams results into a `SnowflakeData` object.

### Execute Custom SQL Queries

For complex analytics, use `SnowflakeIngestor.ingest_query` with parameterized statements:

```python
with SnowflakeIngestor(
    account="myaccount",
    user="myuser",
    password="mypassword",
    warehouse="COMPUTE_WH",
) as ingestor:
    query = """
        SELECT 
            CUSTOMER_ID,
            FIRST_NAME,
            LAST_NAME,
            TOTAL_SPENT
        FROM SALES 
        WHERE DATE > %(date)s 
        AND REGION = %(region)s
    """
    params = {"date": "2024-01-01", "region": "NORTH_AMERICA"}
    data = ingestor.ingest_query(query, params=params)

print(f"Fetched {data.row_count} rows from SALES")

```

The context manager (`__enter__`/`__exit__`) ensures automatic connection cleanup as implemented at lines 647+ in the source file.

## Converting Snowflake Data to Semantica Documents

The critical bridge between Snowflake and semantic processing is `SnowflakeIngestor.export_as_documents`. This method transforms database rows into document structures suitable for vector stores and LLM pipelines:

```python
documents = ingestor.export_as_documents(
    data,
    id_field="CUSTOMER_ID",
    text_fields=["FIRST_NAME", "LAST_NAME", "ADDRESS"],
    metadata_fields=["TOTAL_SPENT", "CREATED_AT", "REGION"]
)

# Documents are dictionaries with id, text, and metadata keys

for doc in documents[:3]:
    print(f"ID: {doc['id']}")
    print(f"Text: {doc['text'][:100]}...")
    print(f"Metadata: {doc['metadata']}")
    print()

```

Each document contains:
- **id**: Unique identifier from your specified `id_field`
- **text**: Concatenated content from `text_fields`
- **metadata**: Dictionary of additional column values for filtering and retrieval

## Schema Introspection

Discover available tables programmatically with `list_tables`:

```python
tables = ingestor.list_tables(database="MYDB", schema="PUBLIC")
print(f"Tables in PUBLIC schema: {tables}")

```

This enables dynamic pipeline construction based on warehouse schema.

## Complete Integration Example

```python
from semantica.ingest import SnowflakeIngestor

def ingest_customer_data():
    """End-to-end Snowflake to Semantica document pipeline."""
    
    with SnowflakeIngestor() as ingestor:  # Credentials from environment

        # Step 1: Verify table exists

        tables = ingestor.list_tables(database="CRM", schema="CUSTOMERS")
        if "VIP_CUSTOMERS" not in tables:
            raise ValueError("Target table not found")
        
        # Step 2: Ingest with business logic filter

        data = ingestor.ingest_table(
            table="VIP_CUSTOMERS",
            where="LIFETIME_VALUE > 10000",
            limit=10000
        )
        
        # Step 3: Convert to documents for embedding

        documents = ingestor.export_as_documents(
            data,
            id_field="CUSTOMER_ID",
            text_fields=["FIRST_NAME", "LAST_NAME", "COMPANY", "NOTES"],
            metadata_fields=["LIFETIME_VALUE", "INDUSTRY", "LAST_CONTACT"]
        )
        
        return documents

# Documents ready for: vector store insertion, LLM context, semantic search

docs = ingest_customer_data()

```

## Key Source Files

| File | Purpose |
|------|---------|
| [`semantica/ingest/snowflake_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/semantica/ingest/snowflake_ingestor.py) | Core implementation: `SnowflakeConnector`, `SnowflakeIngestor`, `SnowflakeData` |
| [`tests/test_snowflake_ingestor.py`](https://github.com/semantica-agi/semantica/blob/main/tests/test_snowflake_ingestor.py) | Unit tests for connection handling, query execution, and edge cases |
| [`cookbook/advanced/snowflake_ingestion_examples.py`](https://github.com/semantica-agi/semantica/blob/main/cookbook/advanced/snowflake_ingestion_examples.py) | Production-ready patterns for batch loading and schema introspection |

## Summary

- **Install** the Snowflake connector with `pip install snowflake-connector-python`
- **Authenticate** via password, OAuth, or key-pair through `SnowflakeIngestor` initialization
- **Ingest data** using `ingest_table` for full tables or `ingest_query` for custom SQL
- **Convert to documents** with `export_as_documents` to generate `id`, `text`, and `metadata` structures
- **Use context managers** (`with` statements) for automatic connection cleanup

## Frequently Asked Questions

### What authentication methods does Semantica's Snowflake integration support?

Semantica supports password-based, OAuth, and key-pair (JWT) authentication as implemented in `SnowflakeConnector.connect`. Pass credentials directly to the constructor or set standard `SNOWFLAKE_*` environment variables. The connector automatically detects the authentication strategy based on provided parameters.

### How does `ingest_table` handle large tables?

The `ingest_table` method applies server-side `LIMIT` and `OFFSET` clauses to paginate results without loading entire tables into memory. For very large datasets, implement client-side pagination by calling `ingest_table` repeatedly with increasing `offset` values, or use `ingest_query` with incremental filters on indexed timestamp columns.

### Can I use parameterized queries with the Snowflake integration?

Yes. The `ingest_query` method accepts a `params` dictionary that uses Snowflake's native parameter binding (`%(key)s` syntax). This prevents SQL injection and handles proper type serialization for dates, timestamps, and numeric values.

### What format does `export_as_documents` produce?

The method returns a list of dictionaries, each containing `id` (string), `text` (concatenated string from specified fields), and `metadata` (dictionary of additional columns). This format is compatible with Semantica's embedding pipelines, vector store loaders, and LangChain document interfaces.