Connections and Hooks: Talking to the Outside World

Your DAG needs to reach a database, an API, a bucket — without a password ever appearing in your code. Connections and hooks are how.

A pipeline that only moves data between Python variables isn’t a pipeline. The moment the bookshop DAG needs to fetch orders and load them somewhere, it has to reach outside the process — to Postgres, to an HTTP endpoint, to a bucket — and every one of those handshakes needs a host and a credential. The wrong instinct is to paste them into the DAG. Airflow gives you two things so you never do that: connections and hooks.

A connection is a named credential

A connection is a stored, encrypted bundle of everything needed to reach one system — host, port, login, password, schema, and a bag of extras for whatever else that system wants. It lives in Airflow’s metadata store (encrypted with your Fernet key), and it has an id. Your DAG refers to the id; the secrets stay out of the code entirely.

There are three ways to create one, and they exist for three different stages of a project’s life.

In the UI. Admin → Connections → add. You fill in a form, pick a connection type, and save. Good for getting started and for connections a human manages.

As an environment variable, in URI form. Airflow reads any AIRFLOW_CONN_<ID> variable and treats it as a connection — the id is the suffix, lowercased:

export AIRFLOW_CONN_BOOKSHOP_DB='postgresql://bookshop:[email protected]:5432/orders'

That defines a connection with id bookshop_db. No UI, no database row — perfect for local dev and containers, where you set it in your Astro project’s .env and it’s just there. The URI packs scheme, login, password, host, port, and path into one string.

A secrets backend, for production. You point Airflow at Vault, AWS Secrets Manager, or GCP Secret Manager, and it looks up connections there first. Rotation, audit, and access control become someone else’s well-solved problem instead of rows in your Airflow database. Same conn_id in the DAG — only the storage changes underneath. We’ll sketch the config at the end of this chapter and go deep in a later one.

Whichever you pick, the rule is absolute: a password never appears in a DAG file. If you can grep a secret out of your repo, you’ve already lost.

The connection URI, in full

That one-line URI hides more structure than it looks. It’s worth pulling apart, because the URI is the canonical, portable form of a connection — it’s what an env var holds, what the CLI emits, and what you’ll paste into a teammate’s .env when they can’t get the API to respond. The full grammar is:

<conn_type>://<login>:<password>@<host>:<port>/<schema>?<extra as query params>

Every field is optional except the scheme. A bare http://api.internal is a valid connection with only a host. But two parts trip people up, and both are about encoding.

Passwords must be percent-encoded. The URI uses @, :, /, and ? as structural characters, so if your password contains any of them, the parser will misread where the password ends and the host begins. A password of p@ss/w0rd is not p@ss/w0rd in a URI — it’s a broken URI. Encode it:

# password is literally:  p@ss/w0rd
export AIRFLOW_CONN_BOOKSHOP_DB='postgresql://bookshop:p%40ss%[email protected]:5432/orders'

%40 is @, %2F is /. When you construct a URI by hand and the connection mysteriously points at the wrong host, this is almost always why. If you’re building the URI in Python, don’t guess — let the machine encode it:

from urllib.parse import quote_plus

pw = quote_plus("p@ss/w0rd")   # -> 'p%40ss%2Fw0rd'
uri = f"postgresql://bookshop:{pw}@db.internal:5432/orders"

The extra bag rides in the query string. Everything after the ? becomes key-value pairs in the connection’s extra field — the place per-provider config lives. ?sslmode=require&connect_timeout=10 sets two extras. Because these are query params, their values get percent-encoded too, which is why a hand-typed options=-c search_path=orders needs its spaces encoded.

Going the other direction is easy, and you’ll do it constantly to verify what Airflow actually stored. Every Connection object can render itself back to a URI, and the CLI wraps that:

airflow connections get bookshop_db --output json
airflow connections export connections.env --file-format env --serialization-format uri

Or in Python, when you’re debugging a DAG that connects to the wrong place:

from airflow.sdk import BaseHook

conn = BaseHook.get_connection("bookshop_db")
print(conn.get_uri())   # the exact URI Airflow resolved, password and all

get_uri() round-trips: the URI it prints will reconstruct the same connection. That symmetry is the whole reason URIs are the interchange format — you can move a connection between environments as a single string and know nothing was lost.

The extra field is where providers get specific

Host, login, password, port, schema — five fields cover the shape of a connection, but no real system is that simple. Postgres wants an SSL mode. Snowflake wants a warehouse and a role. An S3 endpoint wants a region and maybe a custom endpoint URL. Airflow’s answer to “where does all that go” is the extra field: a free-form JSON blob attached to the connection, which every hook reads through the parsed property extra_dejson.

Here’s the bookshop’s Postgres connection with a realistic extra, built as a URI:

export AIRFLOW_CONN_BOOKSHOP_DB='postgresql://bookshop:[email protected]:5432/orders?sslmode=require&connect_timeout=10&options=-c%20search_path%3Dorders'

Or, because JSON is easier to read than a query string, the same connection defined the way the UI and airflow_settings.yaml store it:

{
  "conn_type": "postgres",
  "host": "db.internal",
  "login": "bookshop",
  "password": "s3cr3t",
  "schema": "orders",
  "port": 5432,
  "extra": {
    "sslmode": "require",
    "connect_timeout": 10,
    "options": "-c search_path=orders"
  }
}

PostgresHook reads those extra keys and passes them straight through to the underlying driver when it opens a connection — sslmode=require becomes a real TLS requirement, connect_timeout bounds the dial, options sets the session search path. You didn’t write a line of driver code; you described the connection and the hook honored it.

In your own tasks you can read the same bag directly:

conn = BaseHook.get_connection("bookshop_db")
if conn.extra_dejson.get("sslmode") != "require":
    raise RuntimeError("refusing to connect to prod without TLS")

The lesson: when a provider’s docs mention a config knob that isn’t host/port/login/password/schema, it almost certainly belongs in extra. Warehouse and role for Snowflake, region_name for AWS, verify for TLS trust — all of them live there, out of your DAG, resolved at connect time.

The pattern generalizes to every provider, which is why it’s worth getting comfortable with. A Snowflake connection, for instance, has no field for “warehouse” or “role” in the five core slots — because those are Snowflake-specific, they ride in extra:

{
  "conn_type": "snowflake",
  "login": "bookshop_loader",
  "password": "s3cr3t",
  "schema": "orders",
  "extra": {
    "account": "ab12345.us-east-1",
    "warehouse": "LOAD_WH",
    "role": "TRANSFORMER",
    "database": "BOOKSHOP"
  }
}

SnowflakeHook reads account, warehouse, and role from extra and threads them into the driver’s session — same mechanism as Postgres’ sslmode, different keys. Once you’ve seen two providers do it, you can predict how the third will: read the provider’s connection docs, find the extra keys, put them in the bag.

A hook is the client that uses it

A connection is inert data. A hook is the thing that turns that data into a live, ready-to-use client. You hand a hook a conn_id, it looks the connection up, wires the credentials, and hands you back an object you can actually call — a database cursor, an HTTP session, a bucket client — with none of the connect-and-authenticate boilerplate.

Every integration provider ships hooks. PostgresHook for Postgres, HttpHook for HTTP APIs, S3Hook for S3-compatible storage, and hundreds more. The shape is always the same: construct it with a conn_id, get a configured client.

Here’s the bookshop’s daily load as a TaskFlow task that grabs a PostgresHook and writes the day’s orders:

from airflow.sdk import task
from airflow.providers.postgres.hooks.postgres import PostgresHook

@task
def load_orders(rows, data_interval_start):
    hook = PostgresHook(postgres_conn_id="bookshop_db")
    day = data_interval_start.strftime("%Y-%m-%d")
    hook.run(
        "DELETE FROM orders WHERE order_date = %s",
        parameters=(day,),
    )
    hook.insert_rows(table="orders", rows=rows)

Notice what isn’t there: no host, no password, no psycopg2.connect(...). The hook resolved all of it from bookshop_db. Notice also the DELETE before the insert — that’s the idempotency rule from the scheduling post, made concrete. Run this task twice for the same day and you get one day’s orders, not two.

Hooks do more than .run()

hook.run(sql) is the workhorse for statements you don’t need results from — DDL, DELETE, INSERT. But a database hook is a small, well-stocked toolbox, and reaching for the right method saves you from writing cursor plumbing by hand. On any DB-API hook you get:

hook = PostgresHook(postgres_conn_id="bookshop_db")

# a list of tuples — every matching row
rows = hook.get_records("SELECT id, title FROM books WHERE in_stock")

# just the first row (or None) — perfect for scalars and existence checks
(count,) = hook.get_first("SELECT count(*) FROM orders WHERE order_date = %s",
                          parameters=(day,))

# a pandas DataFrame, when the next step is analysis
df = hook.get_pandas_df("SELECT * FROM orders WHERE order_date = %s",
                        parameters=(day,))

get_records and get_first are your everyday reads; get_pandas_df is the shortcut when a task’s job is to pull a slice and reshape it. All three accept parameters for safe, driver-side substitution — never build SQL with f-strings around user data.

Two more methods matter when a library wants a connection you didn’t get from Airflow. get_conn() returns the raw DB-API connection — the actual psycopg2 connection object — so you can manage transactions yourself, stream a cursor, or hand it to code that expects a plain connection:

conn = hook.get_conn()
with conn.cursor() as cur:
    cur.execute("SELECT id FROM orders WHERE total > %s", (100,))
    for (order_id,) in cur:      # stream, don't materialize
        ...
    conn.commit()

And get_sqlalchemy_engine() hands you a SQLAlchemy engine wired to the same connection — which is exactly what pandas.to_sql, SQLModel, or dbt-adjacent tooling want:

engine = hook.get_sqlalchemy_engine()
df.to_sql("orders", engine, if_exists="append", index=False)

The through-line: whatever the downstream library expects — a cursor, a raw connection, an engine, a DataFrame — the hook can produce it from the same conn_id, so the credential never leaves Airflow’s resolution path.

One habit that pays off: construct a hook once and reuse it across queries within a task, rather than newing one up per statement. Each fresh PostgresHook(...) plus its first .run() re-resolves the connection and opens a socket; a hook you hold onto can reuse its connection for a batch of reads. And when you’re inserting many rows, prefer insert_rows(..., commit_every=1000) over a loop of single inserts — it batches and commits in chunks, which is the difference between a load that finishes in seconds and one that hammers the database a row at a time.

Underneath all of this is BaseHook.get_connection(conn_id), which every hook calls to fetch the Connection object. You can call it yourself when you need a field a hook doesn’t expose — pulling a token out of extra to build a bespoke client, say. It’s the one method that’s identical across every provider, and it’s your escape hatch when no packaged hook fits.

HTTP is a hook too

Fetching from the mock orders API is the same idea with a different hook:

from airflow.providers.http.hooks.http import HttpHook

@task
def fetch_orders(data_interval_start):
    hook = HttpHook(method="GET", http_conn_id="orders_api")
    day = data_interval_start.strftime("%Y-%m-%d")
    resp = hook.run(endpoint=f"/orders/{day}.json")
    resp.raise_for_status()
    return resp.json()

The orders_api connection holds the base URL in its host field and any credentials to match. HttpHook reads the connection’s login/password for basic auth automatically, and you can set auth_type (for token or digest schemes) or drop a bearer token into extra as a default header. Swap the endpoint from a mock to production by editing the connection — the DAG doesn’t change a line.

HttpHook also does the thing you’d otherwise hand-roll: retry with backoff. run_with_advanced_retry wraps a request in a tenacity retry policy, so a flaky upstream gets a few polite re-tries before the task fails:

import tenacity

resp = hook.run_with_advanced_retry(
    endpoint=f"/orders/{day}.json",
    _retry_args={
        "stop": tenacity.stop_after_attempt(4),
        "wait": tenacity.wait_exponential(multiplier=1),
    },
)

_retry_args is handed straight to tenacity.Retrying(**_retry_args), so it takes tenacity objects, not loose keywords. You will find plenty of snippets online passing {"stop_after_attempt": 4, "wait_exponential_multiplier": 1000} — that shape is Airflow 1.x-era, and on a current tenacity it dies with TypeError: BaseRetrying.__init__() got an unexpected keyword argument 'stop_after_attempt'. If a retry helper blows up on a kwarg you didn’t invent, this is usually why.

One newcomer snag worth naming: for an HTTP connection, the base URL is assembled from the connection’s host (and optional schema, which here means the scheme — http or https — not a database schema), and the endpoint you pass to .run() is appended to it. So a connection with host=api.internal and schema=https, called with endpoint="/orders/2026-06-14.json", hits https://api.internal/orders/2026-06-14.json. Get the split wrong — a full URL in host, or the scheme baked into the endpoint — and you’ll chase 404s that are really malformed URLs.

That’s the pattern across every hook family — the packaged client already knows how to be a good network citizen, so you don’t sprinkle try/except and time.sleep through your DAG.

Variables are for knobs, not keys

Airflow has a second store that looks deceptively similar: Variables. A Variable is a named piece of config — a batch size, a feature flag, a target table name, a list of regions to process. You read and write them through the Task SDK:

from airflow.sdk import Variable

Variable.set("bookshop_batch_size", 500)
batch_size = Variable.get("bookshop_batch_size", default=500)

Variables shine when the value is structured. Store JSON and ask for it back parsed:

Variable.set("bookshop_regions", ["us", "eu", "apac"], serialize_json=True)
regions = Variable.get("bookshop_regions", deserialize_json=True)   # a real list

Like connections, Variables have an env-var shortcut — AIRFLOW_VAR_<NAME> (uppercase name, value as a string). Setting AIRFLOW_VAR_BOOKSHOP_BATCH_SIZE=500 in your .env gives you the same Variable without a database row, which is how you keep local config out of the metastore.

The distinction that matters: Variables are for non-secret configuration; Connections are for credentials and endpoints. A Variable holds a tuning knob you might change without touching code. A Connection holds the thing you’d be fired for leaking. Don’t stuff a database password into a Variable because it’s convenient — Variables aren’t built to be the encrypted, access-controlled, rotatable store that Connections and secrets backends are. (You can serve Variables from a secrets backend when they hold something sensitive-ish, but the mental split holds: knobs versus keys.)

The one Variable mistake everyone makes

There’s a trap in how Variables get read, and it’s worth flagging loudly because it’s silent until it isn’t. Never call Variable.get() at the top level of a DAG file:

# DON'T — this runs on every parse
BATCH = Variable.get("bookshop_batch_size", default=500)

@dag(schedule="@daily", start_date=...)
def bookshop():
    ...

The scheduler parses every DAG file repeatedly — many times a minute — to notice changes. Any code at the top level runs on every parse, and Variable.get() at the top level means a database round-trip on every parse, for every worker and the scheduler both. Multiply by a folder of DAGs and you’ve made your config lookups a scheduler bottleneck. Read Variables inside a task, where the code runs once per execution, not once per parse — or use templating, which defers the lookup to render time:

@task
def load(batch_size):
    ...

load(batch_size="{{ var.value.bookshop_batch_size }}")

Templating: reach config without importing anything

Both stores are exposed to Jinja templates, which is often the cleanest way to feed config into an operator without a line of Python. In any templated field you have:

{{ var.value.bookshop_batch_size }}     # a Variable, as a string
{{ var.json.bookshop_regions[0] }}      # a JSON Variable, parsed and indexed
{{ conn.bookshop_db.host }}             # a Connection's fields, one at a time
{{ conn.bookshop_db.login }}
{{ conn.bookshop_db.schema }}

{{ conn.bookshop_db.host }} is handy for building a log line or a command string that needs the host but not the password — you name the resource and Jinja pulls the field at render time, so nothing is hardcoded and nothing secret leaks into the template. Because these resolve during task execution (not at parse), they don’t carry the top-level penalty that Variable.get() does.

Managing connections and variables from the CLI

Clicking through the UI is fine for one connection. For a project you check into git, you want connections and variables defined as data you can version, review, and replay. The CLI is built for exactly this:

# add a connection from a URI
airflow connections add bookshop_db \
  --conn-uri 'postgresql://bookshop:[email protected]:5432/orders?sslmode=require'

# inspect and list
airflow connections get bookshop_db --output json
airflow connections list

# round-trip a whole environment's connections to a file, and back
airflow connections export connections.json
airflow connections import connections.json

Variables mirror this with airflow variables set/get/list/export/import. The export/import pair is the one to remember: it’s how you snapshot a working environment and rehydrate it somewhere else, and it’s diff-able in a pull request.

If you’re on the Astro CLI — the local stack this series uses — there’s an even smoother path. You declare connections and variables in airflow_settings.yaml at the root of your project:

airflow:
  connections:
    - conn_id: bookshop_db
      conn_type: postgres
      conn_host: db.internal
      conn_schema: orders
      conn_login: bookshop
      conn_password: s3cr3t
      conn_port: 5432
      conn_extra: '{"sslmode": "require"}'
  variables:
    - variable_name: bookshop_batch_size
      variable_value: "500"

astro dev start loads that file automatically when the stack comes up, and astro dev object import re-imports it into a running environment after you edit it — no clicking, no re-typing. Keep real secrets out of a committed airflow_settings.yaml (use .env and env-var connections for those); the file is best for the structure of your dev connections, with placeholder credentials.

Secrets backends, at a glance

Everything so far stores config in the metastore or in env vars. In production you usually want a third option: a dedicated secrets backend — HashiCorp Vault, AWS Secrets Manager, GCP Secret Manager — that owns your credentials and gives you rotation, audit trails, and access control for free. You wire one up with two settings:

AIRFLOW__SECRETS__BACKEND='airflow.providers.amazon.aws.secrets.secrets_manager.SecretsManagerBackend'
AIRFLOW__SECRETS__BACKEND_KWARGS='{"connections_prefix": "airflow/connections", "variables_prefix": "airflow/variables"}'

The one thing to internalize now is the resolution order. When a hook asks for bookshop_db, Airflow looks in a fixed sequence and takes the first hit:

  1. Environment variables (AIRFLOW_CONN_BOOKSHOP_DB) — checked first, always.
  2. The configured secrets backend — Vault, Secrets Manager, and friends.
  3. The metastore — the database rows the UI writes.

So an env var shadows a backend, and a backend shadows the metastore. That ordering is a feature: it lets a developer override a production connection locally with a single .env line, and it means a misplaced env var can silently mask the “real” credential — worth knowing when a connection resolves to something you didn’t expect. The backend class, backend_kwargs, per-type prefixes, and how this order plays out across deployments get the full treatment in Connections, Variables, and Secrets Backends later in the series; for now, just hold the order in your head: env → backend → metastore.

A word on the “Test” button

The UI’s connection form has a Test button, and in Airflow 3.x it’s disabled by default — deliberately. Testing a connection means the server actually dials the external system using stored credentials, which is a small but real attack surface (and it can trigger side effects on the far end). To turn it on you set AIRFLOW__CORE__TEST_CONNECTION to Enabled (or Hidden to keep it off and out of the UI). When enabled, it calls the hook’s test_connection() and reports success or the actual error, which is a genuine time-saver while you’re wiring things up. Enable it in dev if you like; think twice before enabling it on a shared production instance.

Final thoughts

The payoff of this split is that your DAG becomes portable. The same load_orders task runs against a throwaway Postgres in your Astro dev stack and against the production warehouse, and the only difference is which bookshop_db connection resolves — a .env line locally, a Secrets Manager entry in prod. Everything specific to a system — its TLS mode, its region, its retry policy — lives in the connection’s extra and the hook’s methods, not in your source. Code that names a resource instead of embedding it is code you can move, test, and hand off without a scavenger hunt for hardcoded hosts. Connections and hooks aren’t ceremony. They’re the seam that keeps your credentials out of git and your pipeline honest about the fact that the outside world is configuration, not source.

Next: Reading the UI When a DAG Goes Red

Comments