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

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

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

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:

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:

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:

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:

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:

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:

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

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	Core implementation: SnowflakeConnector, SnowflakeIngestor, SnowflakeData
tests/test_snowflake_ingestor.py	Unit tests for connection handling, query execution, and edge cases
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.

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