Can Semantica Ingest Data from Snowflake? A Complete Technical Guide

Yes, Semantica can ingest data from Snowflake through a dedicated ingestion subsystem located in semantica/ingest/snowflake_ingestor.py, which provides authentication, query execution, and document export capabilities.

The open-source framework Semantica (available at semantica-agi/semantica) includes native support for Snowflake data ingestion. Whether you need to sync entire tables, execute custom SQL queries, or export results into Semantica's document pipeline, the platform offers a robust Python API that handles connection management, type conversion, and progress tracking automatically.

Snowflake Ingestion Architecture

The Snowflake integration in Semantica follows a modular design with three core components defined in semantica/ingest/snowflake_ingestor.py:

SnowflakeConnector: Connection Management

The SnowflakeConnector class manages low-level connections using the official snowflake.connector library. It supports password, key-pair, OAuth, and SSO authentication methods. The class provides test_connection() for validation and disconnect() for resource cleanup. When using key-pair authentication, it automatically loads the private key using the cryptography library.

SnowflakeIngestor: High-Level Data Retrieval

The SnowflakeIngestor class provides the high-level API for data retrieval. This component uses the connector to ingest entire tables, run arbitrary queries, list tables, and fetch schema information. All operations return standardized SnowflakeData objects, and the class integrates with Semantica's get_progress_tracker() for operational visibility.

SnowflakeData: Result Container

The SnowflakeData dataclass serves as a container for query results, holding rows, column names, metadata, and timestamps. It includes the _convert_rows() method to transform Snowflake-specific types (such as datetime and bytes) into JSON-serializable values, ensuring compatibility with downstream processing pipelines.

How Snowflake Data Ingestion Works

The ingestion flow implemented in the source code follows these seven stages:

1. Configuration – Supply credentials directly to constructors or via environment variables (SNOWFLAKE_ACCOUNT, SNOWFLAKE_USER, SNOWFLAKE_PASSWORD, etc.).

2. Connector Initialization – SnowflakeConnector validates required fields and prepares the authentication method. If private_key_path is provided, it loads the key using the cryptography library.

3. Connection Establishment – Calling connector.connect() creates a Snowflake session, runs a lightweight version check, and returns a SnowflakeConnection object.

4. Data Retrieval – SnowflakeIngestor.ingest_table() builds a safe, escaped SELECT statement with quoted identifiers and bound parameters. It supports optional WHERE, ORDER BY, LIMIT, and OFFSET clauses.

5. Query Execution – The system uses a DictCursor to fetch rows and extract column metadata from the Snowflake connection.

6. Result Wrapping – Rows pass through _convert_rows() for type normalization, then pack into a SnowflakeData instance with row counts and column lists.

7. Optional Export – Calling export_as_documents() transforms results into document dictionaries suitable for downstream Semantica pipelines, such as vector-store indexing.

All steps integrate with Semantica's get_logger() utility for consistent logging and error handling.

Practical Code Examples for Snowflake Ingestion

Basic Table Ingestion

from semantica.ingest import SnowflakeIngestor

# Create an ingestor – credentials can also be supplied via env vars

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

# Ingest the first 5,000 rows from the CUSTOMERS table

data = ingestor.ingest_table("CUSTOMERS", limit=5000)

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

Running a Custom Query with Parameters

query = """
SELECT *
FROM SALES
WHERE DATE > %(start_date)s
ORDER BY AMOUNT DESC
LIMIT %(limit)s
"""
params = {"start_date": "2024-01-01", "limit": 1000}

query_data = ingestor.ingest_query(query, params=params)

for row in query_data.data[:3]:
    print(row)

Exporting Snowflake Rows as Semantica Documents

documents = ingestor.export_as_documents(data, text_fields=["NAME", "DESCRIPTION"])
for doc in documents[:2]:
    print(doc["id"], doc["text"])

Listing Tables and Fetching Schema

tables = ingestor.list_tables()
print("Available tables:", tables)

schema_info = ingestor.get_table_schema("CUSTOMERS")
print(schema_info)

Using the Ingestor as a Context Manager

with SnowflakeIngestor(account="myaccount", user="myuser", password="pwd") as ing:
    customers = ing.ingest_table("CUSTOMERS")
    # No explicit disconnect needed – the context manager closes the connection

Summary

• Semantica ingests Snowflake data through semantica/ingest/snowflake_ingestor.py using three main classes: SnowflakeConnector, SnowflakeIngestor, and SnowflakeData.
• The system supports password, key-pair, OAuth, and SSO authentication methods with automatic credential validation.
• You can ingest entire tables via ingest_table() or execute custom SQL with parameterized queries using ingest_query().
• Results automatically convert to JSON-serializable formats via _convert_rows() and can export as documents for vector-store indexing through export_as_documents().
• Built-in progress tracking via get_progress_tracker() and logging via get_logger() provide full operational visibility.

Frequently Asked Questions

Does Semantica support OAuth authentication for Snowflake?

Yes. The SnowflakeConnector class in semantica/ingest/snowflake_ingestor.py supports password, key-pair, OAuth, and SSO authentication methods. You can configure these through constructor parameters or environment variables such as SNOWFLAKE_ACCOUNT and SNOWFLAKE_USER.

Can I ingest specific rows using SQL filters?

Yes. Use the ingest_query() method to execute arbitrary SQL with parameterized filters. Alternatively, ingest_table() accepts optional where_clause, order_by, limit, and offset parameters for table-specific filtering without writing raw SQL.

What data formats does Snowflake ingestion return?

Data returns as a SnowflakeData dataclass containing rows, column names, and metadata. The internal _convert_rows() method automatically transforms Snowflake-specific types like datetime and bytes into JSON-serializable values for downstream processing.

Is there built-in progress tracking for large Snowflake queries?

Yes. All major ingestion steps integrate with Semantica's get_progress_tracker() utility. The system logs progress through get_logger() and provides visibility into connection establishment, query execution, and data conversion phases.

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