Connecting Airflow to Snowflake

Before Cosmos can run dbt on Snowflake, Airflow itself needs a way in — one connection, its own identity, key-pair auth, and a secrets backend that resolves it at runtime.

dbt can log in to Snowflake now. Airflow can’t — and it needs its own way in, because Cosmos runs dbt from inside Airflow, and there are pipeline steps that never touch dbt at all. A COPY INTO to load raw files, a warehouse resize before a heavy model, a one-off CALL to a stored procedure: those are Airflow talking straight to Snowflake, and they need a connection that exists whether or not dbt is involved.

So this post is short on ceremony and long on the two things that actually bite people: who Airflow logs in as, and where that credential lives so it never lands in a DAG file. Install the provider, give Airflow its own identity, wire the connection through a secrets backend that resolves at runtime, and prove the whole thing with a query. By the end, the non-dbt tasks the series keeps promising are ordinary operators on a connection you trust.

The provider, and the version that matters

Snowflake support isn’t in core Airflow; it’s a provider package. Pin it in requirements.txt, and pin it to a version that isn’t older than 6.3.0:

apache-airflow-providers-snowflake>=6.3.0

The version floor isn’t superstition. Before 6.3.0 the provider accepted an inline private key in the connection as raw PEM text, which is a nightmare to stuff into an environment variable or a secrets-manager value — PEM is multi-line, and every layer between you and Airflow (shell, JSON, a Vault UI) mangles newlines differently. From 6.3.0 the connection’s private_key_content field is base64-encoded, so the key becomes one long single-line token that survives being pasted into JSON, exported as an env var, or stored as a secret without a single escaped newline. If you’re going to hold the key inline — and with a secrets backend you will — 6.3.0 is the version where that stops being painful.

The provider gives you three things: the snowflake connection type, the SnowflakeHook, and — through the common-SQL provider it depends on — the operators for running SQL against Snowflake from a task.

Two service users, not one

Post 2 created svc_dbt. The temptation now is to reuse it for Airflow and be done. Resist it. Give Airflow its own Snowflake user, svc_airflow, with its own key pair, and the reasons compound quickly.

The first is audit. Snowflake’s QUERY_HISTORY records the user behind every statement. If dbt and Airflow share one login, you cannot tell, from Snowflake’s side, whether a runaway query on the warehouse came from a dbt model or from a hand-written COPY INTO in a DAG — they’re the same USER_NAME. Split them and the account usage views become a ledger: svc_dbt ran the transforms, svc_airflow ran the loads and the maintenance. When something burns credits at 3 a.m., the query history tells you which tool to go read.

The second is least privilege. dbt reads raw data and writes models; that’s its whole job, and transformer is scoped to exactly that. Airflow’s non-dbt work is different in kind — it issues COPY INTO against ingestion stages, it resizes warehouses, it calls operational procedures. Those need grants dbt should never hold. Model them as a separate role:

CREATE USER svc_airflow
  DEFAULT_ROLE = airflow_ops
  DEFAULT_WAREHOUSE = ingest_wh;

ALTER USER svc_airflow SET RSA_PUBLIC_KEY = 'MIIBIjANBgkq...IDAQAB';

CREATE ROLE airflow_ops;
GRANT ROLE airflow_ops TO USER svc_airflow;

-- what Airflow's own tasks need, and dbt does not
GRANT OPERATE, MODIFY ON WAREHOUSE ingest_wh TO ROLE airflow_ops;
GRANT USAGE ON DATABASE raw TO ROLE airflow_ops;
GRANT CREATE TABLE, INSERT ON SCHEMA raw.landing TO ROLE airflow_ops;

Same key-pair mechanism as post 2 — generate an RSA pair with OpenSSL, register the public half with ALTER USER ... SET RSA_PUBLIC_KEY, hand Airflow the private half. Nothing about the login is a shared secret on the wire; Airflow signs a token, Snowflake verifies it with the public key. The only thing that changed is that this is a second identity, with a second key, scoped to a role dbt will never assume.

When Cosmos runs dbt in the next post, it can still use svc_dbt through its own connection, or run dbt under svc_airflow if you’d rather have one orchestration identity — that’s a policy call. But for Airflow’s own SQL, svc_airflow is the identity, and keeping it distinct is the cheap version of an audit trail you’ll wish you had later.

One connection, and which field goes where

Airflow keeps credentials in named connections rather than in DAG code. You create a connection of type snowflake, give it a conn_id, and every Snowflake task points at it by that id. The subtlety with the Snowflake provider is that the connection has a handful of top-level fields and an open-ended extra blob, and it matters which value goes where:

  • Login (top-level) — the service user, svc_airflow.
  • Password (top-level) — not a Snowflake password. This field holds the passphrase for the encrypted private key. Key-pair auth still needs a secret to unlock the .p8, and this is where it goes.
  • Schema (top-level) — the default schema, if you want one.
  • Extra (JSON blob) — everything else the session needs: account, warehouse, database, role, and the key itself as private_key_file (a path) or private_key_content (base64, inline).

The warehouse, role, and database live in extra because the provider treats them as session defaults rather than first-class connection identity — which is exactly right, because a single connection often runs tasks that want different warehouses, and you’ll override the warehouse per-task later without touching the stored connection.

You can create it from the CLI so it’s reproducible rather than clicked into a form:

airflow connections add snowflake_default \
  --conn-type snowflake \
  --conn-login svc_airflow \
  --conn-password "$SNOWFLAKE_KEY_PASSPHRASE" \
  --conn-extra '{
    "account": "MYORG-MYACCT",
    "warehouse": "INGEST_WH",
    "database": "RAW",
    "role": "AIRFLOW_OPS",
    "private_key_file": "/secure/svc_airflow.p8"
  }'

That works, and for a laptop or a single box it’s fine. But it stores the connection in Airflow’s metadata database, and the passphrase came from your shell history. For anything shared — and certainly for anything in production — the connection shouldn’t live in the metadata DB at all. It should live in a secrets backend, and Airflow should resolve it at runtime.

Where the credential actually lives

Airflow resolves a connection through a chain: a secrets backend first (if one is configured), then environment variables, then the metadata database. That order is the whole game — configure a backend and Airflow reads the connection from there without the credential ever touching the DB or a DAG.

The simplest form is an environment variable, and it’s worth understanding because every backend is a fancier version of it. Airflow reads a connection named X from AIRFLOW_CONN_X (uppercased). So snowflake_default becomes:

export AIRFLOW_CONN_SNOWFLAKE_ANALYTICS='{
  "conn_type": "snowflake",
  "login": "svc_airflow",
  "password": "the-key-passphrase",
  "extra": {
    "account": "MYORG-MYACCT",
    "warehouse": "INGEST_WH",
    "database": "RAW",
    "role": "AIRFLOW_OPS",
    "private_key_content": "LS0tLS1CRUdJTiBFTkNSWVBURUQg...base64...LS0tLS0K"
  }
}'

Two things to notice. First, this is the JSON form — Airflow accepts either a URI (snowflake://svc_airflow:pass@/?account=...&warehouse=...) or JSON, and JSON is far easier to read and to get right than URL-encoding a private key into a query string. Second, this is where private_key_content earns its base64 encoding: the entire .p8 — including its BEGIN ENCRYPTED PRIVATE KEY armor — is base64’d into a single line that drops cleanly into a JSON string. No private_key_file on disk, no newline escaping. That’s the 6.3.0 behavior paying off. Produce the value with:

base64 -w0 /secure/svc_airflow.p8    # one long line, no wrapping

The env-var form is real and useful — it’s how you inject a connection into a CI job or a container without a database — but it’s still a secret sitting in an environment, visible to anything that can read the process. The production answer moves it one layer out, into a managed secrets store, and points AIRFLOW__SECRETS__BACKEND at it.

A secrets backend, concretely

Set two config values (env vars, or the [secrets] block in airflow.cfg): the backend class, and its kwargs. Here’s HashiCorp Vault:

export AIRFLOW__SECRETS__BACKEND=airflow.providers.hashicorp.secrets.vault.VaultBackend
export AIRFLOW__SECRETS__BACKEND_KWARGS='{
  "connections_path": "connections",
  "mount_point": "airflow",
  "url": "https://vault.internal:8200"
}'

With that, when a DAG asks for snowflake_default, Airflow reads the secret at airflow/connections/snowflake_default in Vault. You store the connection there as the same JSON blob you’d have put in the env var — conn_type, login, password, extra with the base64 key — under a conn_uri or JSON value. The DAG names the connection; Vault holds the bytes; the metadata DB never sees the key.

AWS Secrets Manager is the same shape with a different class:

export AIRFLOW__SECRETS__BACKEND=airflow.providers.amazon.aws.secrets.secrets_manager.SecretsManagerBackend
export AIRFLOW__SECRETS__BACKEND_KWARGS='{
  "connections_prefix": "airflow/connections",
  "region_name": "us-east-1"
}'

Now the connection is a secret named airflow/connections/snowflake_default in Secrets Manager, and the workers authenticate to AWS with an IAM role rather than a static credential — so the only long-lived secret in the whole chain is Snowflake’s private key, and it’s encrypted at rest in a store built to hold it. GCP Secret Manager is airflow.providers.google.cloud.secrets.secret_manager.CloudSecretManagerBackend with {"connections_prefix": "airflow-connections", "project_id": "my-project"}, and workers authenticate with a service account. Same pattern, three clouds: the DAG references a conn_id, the backend resolves it, and nothing sensitive is in Airflow’s database or your repo.

A useful property falls out of the resolution order: because the backend is checked before the metadata DB, you can keep a throwaway connection in the DB for local dev and have production transparently override it from Vault — same conn_id, different source, zero DAG changes.

The payoff you feel later is rotation. Key-pair auth is only as good as your ability to replace a key without a redeploy, and Snowflake makes that graceful: a user can hold two public keys at once (RSA_PUBLIC_KEY and RSA_PUBLIC_KEY_2). To rotate svc_airflow, register the new public key in the second slot, update the connection secret in Vault or Secrets Manager to the new base64 private key, let Airflow pick it up on the next task, then unset the old slot. No DAG changes, no image rebuild, no window where the pipeline can’t sign in — because the credential was never baked into anything you’d have to rebuild. Try that when the key is inlined in a DAG or lives in a profiles.yml on the worker, and rotation becomes a deploy with downtime instead of a two-line secret edit.

The smoke test

Prove the connection before you build anything on it. The modern way to run SQL from a task is SQLExecuteQueryOperator, the generic operator that works across every database provider — the old SnowflakeOperator still exists but is deprecated, so reach for the generic one in anything new:

from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

check_connection = SQLExecuteQueryOperator(
    task_id="check_connection",
    conn_id="snowflake_default",
    sql="select current_version(), current_role(), current_warehouse()",
)

A green run means Airflow pulled the connection from the backend, unlocked the key with the passphrase, signed in as svc_airflow, and got an answer back. Returning current_role() and current_warehouse() in the same query is a cheap way to confirm the extra values actually took effect — if the role comes back as something you didn’t expect, your extra didn’t parse.

The operator has a few parameters worth knowing before the non-dbt tasks arrive:

  • hook_params overrides connection defaults per task without editing the stored connection. A task that needs a bigger warehouse for one heavy statement passes hook_params={"warehouse": "BIG_WH"} and leaves everyone else on the default.
  • autocommit defaults to True for the Snowflake hook; set it False when a task issues several statements that must land together or not at all.
  • split_statements lets a single sql string hold multiple ;-separated statements and run them in order — handy for a small setup script, and off by default so a stray semicolon in one statement doesn’t silently fan out into several.
  • parameters binds values into the SQL instead of formatting them in, which is how you keep templated values from becoming an injection hole (more below).

Reading a result back with SnowflakeHook

SQLExecuteQueryOperator is perfect for doing something. When a task needs to decide something — read a row and branch, or assert a count — drop into a SnowflakeHook inside a TaskFlow task and handle the result directly:

from airflow.sdk import task
from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook

@task
def assert_orders_loaded(day: str):
    hook = SnowflakeHook(snowflake_conn_id="snowflake_default")
    (count,) = hook.get_first(
        "select count(*) from raw.landing.orders where o_orderdate = %(day)s",
        parameters={"day": day},
    )
    if count == 0:
        raise ValueError(f"no orders landed for {day}")
    return count

The hook gives you the whole spectrum of result handlers: get_first for one row, get_records for all rows as a list of tuples, get_pandas_df when a downstream step wants a DataFrame, and plain run for statements whose return value you don’t care about. run also accepts a list of SQL strings and an autocommit flag, so a multi-step maintenance task can live in one call.

Notice parameters={"day": day} and the %(day)s placeholder rather than an f-string. This is not a style preference. day here comes from {{ ds }} — Airflow’s run date — and templated values are exactly the kind of thing that ends up user-influenced in a real system. Binding the parameter hands the value to the Snowflake driver as data, so a value like 2026-05-25'; drop table orders; -- is compared as a string and does nothing. Format it into the SQL with an f-string and you’ve written the classic injection. parameters= works identically on SQLExecuteQueryOperator and on every hook method, so there’s no reason to ever build SQL by concatenation.

The non-dbt tasks, for real

Everything above exists to make these three ordinary. Each is a plain operator on snowflake_default, and each is the kind of step that sits around the dbt models in a real pipeline.

Load raw files with COPY INTO. The ingestion step that lands data before dbt ever runs:

load_orders = SQLExecuteQueryOperator(
    task_id="load_orders",
    conn_id="snowflake_default",
    sql="""
        copy into raw.landing.orders
        from @raw.landing.orders_stage
        file_format = (type = 'PARQUET')
        match_by_column_name = case_insensitive
        on_error = 'ABORT_STATEMENT'
    """,
    hook_params={"warehouse": "INGEST_WH"},
)

There’s a quiet reason COPY INTO is the right ingestion primitive to hand Airflow, beyond it being one statement: it’s idempotent by default. Snowflake records which files a stage has already loaded and skips them on a re-run, so a COPY INTO that partially succeeds and then gets retried by Airflow doesn’t double-load the rows it already ingested — it picks up only the files it hasn’t seen. That property is precisely what makes the retries: 3 policy below safe on a load task: without it, a retry after a mid-load network drop would be a data-duplication bug; with it, the retry is a no-op for the completed files and a clean finish for the rest. on_error = 'ABORT_STATEMENT' keeps a single malformed file from silently loading a partial batch, and match_by_column_name decouples the load from column order in the source files. (If you ever do need to reload files Snowflake already remembers — a corrected extract, say — that’s what force = true is for, and it’s a deliberate override, not the default.)

Resize a warehouse before a heavy model, and shrink it after. Snowflake resizes are near-instant and billed by the second, so scaling up for one expensive step and back down afterward is a legitimate cost lever:

scale_up = SQLExecuteQueryOperator(
    task_id="scale_up",
    conn_id="snowflake_default",
    sql="alter warehouse transforming set warehouse_size = 'LARGE'",
)

scale_down = SQLExecuteQueryOperator(
    task_id="scale_down",
    conn_id="snowflake_default",
    sql="alter warehouse transforming set warehouse_size = 'SMALL'",
    trigger_rule="all_done",
)

The all_done trigger rule on scale_down matters: you want the warehouse to shrink even if the heavy model between the two failed, or you’ll leave a LARGE warehouse running on a red pipeline. That grant to MODIFY ON WAREHOUSE back in the svc_airflow role is what makes this statement legal — and a grant svc_dbt deliberately doesn’t have.

Call a stored procedure for an operational task that’s cleaner as SQL than as Python:

refresh_dim_date = SQLExecuteQueryOperator(
    task_id="refresh_dim_date",
    conn_id="snowflake_default",
    sql="call raw.ops.rebuild_date_spine(%(start)s, %(end)s)",
    parameters={"start": "2020-01-01", "end": "2030-12-31"},
)

None of these should run without retry and timeout policy, because all three touch a warehouse and any warehouse call can hit a transient network blip. Set the defaults once on the DAG:

from datetime import datetime, timedelta
from airflow.sdk import DAG

default_args = {
    "retries": 3,
    "retry_delay": timedelta(minutes=2),
    "retry_exponential_backoff": True,
    "execution_timeout": timedelta(minutes=30),
}

with DAG(
    dag_id="tpch_ingest",
    schedule="@daily",
    start_date=datetime(2026, 5, 1),
    catchup=False,
    default_args=default_args,
) as dag:
    scale_up >> load_orders >> refresh_dim_date >> scale_down

retries with exponential backoff rides out transient errors; execution_timeout is the guardrail that matters most on Snowflake, because a query that hangs is a warehouse that keeps billing. A task with no timeout and a stuck statement is an open tab on your credit card — cap it, and a runaway becomes a failed task instead of a surprise on the invoice.

Tag every query so the audit trail is legible

Splitting svc_airflow from svc_dbt gets you who. A query tag gets you which task. Snowflake stamps a session-level QUERY_TAG onto every statement it records, and if you set it to the DAG and task that issued the SQL, QUERY_HISTORY stops being a wall of anonymous statements and becomes a per-task cost-and-latency ledger you can group by.

Set it through the hook’s session_parameters, which you can attach on the connection’s extra for a blanket default or per-task via hook_params for something templated:

load_orders = SQLExecuteQueryOperator(
    task_id="load_orders",
    conn_id="snowflake_default",
    sql="copy into raw.landing.orders from @raw.landing.orders_stage ...",
    hook_params={
        "warehouse": "INGEST_WH",
        "session_parameters": {
            "QUERY_TAG": "airflow:tpch_ingest:load_orders:{{ ds }}",
        },
    },
)

Now a question like “what did the ingest DAG cost us on May 25th?” is a where query_tag like 'airflow:tpch_ingest:%2026-05-25' against snowflake.account_usage.query_history, not a forensic exercise. Combined with the separate svc_airflow identity, you can attribute every credit either tool spent, down to the task and the run — which is the kind of thing nobody sets up until a Snowflake bill forces the question, and then wishes they’d had all along.

Testing the connection in CI

The one failure mode worse than a broken connection is a broken connection you discover at 2 a.m. instead of on a pull request. Airflow ships a command that resolves a connection and actually opens it:

airflow connections test snowflake_default

It runs the connection through the same resolution chain the scheduler uses — backend, then env, then DB — so it proves the secrets wiring, not just that a row exists somewhere. (One gotcha: for safety Airflow disables connection testing from the UI and REST API by default; the CLI connections test still runs, and if you want the UI button too you opt in with AIRFLOW__CORE__TEST_CONNECTION=Enabled. In CI you want the CLI form anyway, so the default is fine.) In CI, export the connection as AIRFLOW_CONN_SNOWFLAKE_ANALYTICS from your pipeline’s own secret store (or point the job at the real backend with a read-only, non-prod identity) and run the test as a step. If the key rotated, the passphrase drifted, or the base64 got truncated on the way into a secret, the pipeline goes red on a branch — where a red build is an inconvenience — instead of during a nightly run, where it’s an incident. Pair it with airflow dags test tpch_ingest 2026-05-25 to execute the whole DAG against a scratch schema, and a bad credential never reaches production.

Final thoughts

The connection looks like a throwaway prerequisite, but it’s the hinge the rest of the stack hangs on. Get it right and three things become true at once: Airflow reaches Snowflake as its own auditable identity, the private key lives in a secrets backend and nowhere near a DAG, and every non-dbt step — loads, resizes, procedure calls — is an ordinary operator you can retry, time out, and test. Cosmos, in the next post, will reuse this exact connection to build dbt’s profile in memory, so the credential you wired once serves both the dbt tasks and Airflow’s own. Configure it as if the key protects your warehouse, because it does — and then stop thinking about it, because you won’t have to.

Next: dbt on Snowflake, Orchestrated by Airflow

Comments