Snowbridge Async

A generic Python asyncio ingestion framework for moving incremental REST and database records into Snowflake. Sources are plugins, loads are natural-key MERGEs, and source watermarks live beside the data in Snowflake—Airflow only schedules one library call.

The S&P 500 pipeline is the reference implementation, not a framework dependency. Its 2-minute-bar workload is approximately 2.0M market records in a 21-session month (500 × 195 × 21), and the checkpointed direct-merge path cut the daily refresh by 70% versus the original S3/Snowpipe workflow.

Architecture

                         data plane
 REST API ─┐       ┌─> bounded record batches ─> temporary stage ─> MERGE
           ├─> async source plugin                                  │
 Database ─┘                                                        │
                                                                    v
 Airflow ─────────> run_job() ─────────────────────────> Snowflake watermark
 (schedule, retry, alert only)

 Snowflake tables ─> semantic view (30 metrics, 3 join paths) ─> MCP server

The important failure boundary is one Snowflake transaction:

1. merge a bounded batch on its declared natural key;
2. update that job's JSON cursor;
3. commit both, or roll back both.

Replaying a run produces the same target rows. Partitioned APIs may merge intermediate batches without moving the watermark, then checkpoint on their final batch; a crash simply replays safe MERGEs.

Framework

Every job combines a source and a target table:

from src.ingestion import (
    Job, RESTSource, RESTSourceConfig, Table, run_job
)
from src.loading import SnowflakeWarehouse

source = RESTSource(RESTSourceConfig(
    url="https://api.example.com/orders",
    records_path="data.items",
    watermark_parameter="updated_after",
    watermark_field="updated_at",
    tie_breaker_field="order_id",
    page_size=5_000,
))

job = Job(
    name="commerce.orders",
    source=source,
    table=Table(
        name="RAW.ORDERS",
        columns=("order_id", "updated_at", "status", "amount"),
        keys=("order_id",),
    ),
)

result = await run_job(job, SnowflakeWarehouse())

DatabaseSource works with any Python DB-API driver. A query factory keeps dialect-specific incremental SQL in the adapter:

from src.ingestion import DatabaseSource

def query(cursor, batch_size):
    since = (cursor or {}).get("updated_at", "1970-01-01")
    last_id = (cursor or {}).get("id", 0)
    return (
        """
        SELECT id, updated_at, payload
        FROM source_events
        WHERE updated_at > ? OR (updated_at = ? AND id > ?)
        ORDER BY updated_at, id
        """,
        (since, since, last_id),
    )

source = DatabaseSource(
    connection_factory=open_source_connection,
    query_factory=query,
    cursor_factory=lambda row: {
        "updated_at": row["updated_at"],
        "id": row["id"],
    },
    batch_size=10_000,
)

The complete ingestion loop is in src/ingestion/core.py. Neither source requires Airflow, S3, or stock-specific code.

Snowflake model

Run sql/snowflake_setup.sql, then sql/semantic_view.sql.

The control schema records:

Table	Purpose
INGESTION_WATERMARKS	one composite JSON cursor per source job

The stock proof case uses stable natural keys:

DIM_SECTOR (sector)
    ↑ company_to_sector
DIM_COMPANY (symbol, sector, ...)
    ↑ bar_to_company
FACT_MARKET_BARS (symbol, observed_at, source_interval, OHLCV, ...)
    └─ bar_to_calendar ─> DIM_DATE (date_day, year, quarter, month, ...)

MARKET_ANALYTICS defines 30 governed metrics—including volume, dollar volume, VWAP, price ranges, intraday returns, volatility, up/down rates, dividends, and observation bounds—and explicit company, sector, and calendar join paths.

Conversational analytics

The MCP server exposes the semantic catalog and a parameterized metric-query tool. It accepts allowlisted metrics, dimensions, and filters; it does not expose arbitrary SQL.

python -m pip install -e ".[mcp]"
python -m mcp_server.server

Example client configuration:

{
  "mcpServers": {
    "market-analytics": {
      "command": "python",
      "args": ["-m", "mcp_server.server"],
      "env": {
        "SNOWFLAKE_ACCOUNT": "...",
        "SNOWFLAKE_USER": "...",
        "SNOWFLAKE_PASSWORD": "..."
      }
    }
  }
}

Example questions:

• “Compare VWAP and total volume by sector this quarter.”
• “Which symbols had the highest intraday volatility last month?”
• “Show the up-bar rate by weekday for 2-minute observations.”

Run the stock proof case

cp .env.example .env
# fill in Snowflake credentials and run both SQL files first
docker compose -f docker/docker-compose.yml up -d

Open Airflow at http://localhost:8080 (admin / admin) and enable generic_stock_market_ingestion. The DAG returns counts and cursors only; data never passes through XCom.

Use MARKET_DATA_INTERVAL=1d for a light demonstration or 2m for the high-volume profile. Intraday retention limits are imposed by the upstream Yahoo Finance service.

Tests

python -m pip install -e ".[mcp,test]"
pytest

Tests cover batch/cursor behavior, replay idempotency, REST and database streaming, identifier safety, generated MERGE SQL, semantic-query allowlists, the 30-metric catalog, and the registered MCP surface.

Earlier pipeline results

These screenshots are retained from the original stock-specific S3/Snowpipe implementation to show the project's evolution.

Airflow DAG

S3 staging

Snowflake fact table