Where are connections actually stored?

In the metadata database, with the password field encrypted using the Fernet key from your Airflow config. Lose that key and every stored password becomes unreadable, which is why the key belongs in your secret manager, not in the repo.

Should I use Variables or environment variables for config?

Environment variables for anything static, since reading a Variable hits the metadata database. Variables are right for values that change without a redeploy, but read them inside a task, never at module level.

What is the difference between a hook and an operator?

A hook is a client — it manages the connection and exposes methods. An operator is a task that usually wraps a hook. Use an operator for the standard case; use the hook directly inside a Python task when you need logic the operator does not offer.

How do I avoid putting secrets in the metadata database at all?

Configure a secrets backend — AWS Secrets Manager, GCP Secret Manager, HashiCorp Vault. Airflow then resolves connections and variables from there at runtime, and nothing sensitive is stored in its own database.

Connections, Hooks, and Variables

A DAG that hardcodes a database password is a DAG you cannot commit. Connections, hooks, and variables are how configuration lives outside the code.

Creating a connection

airflow connections add postgres_analytics \
  --conn-type postgres \
  --conn-host db.internal \
  --conn-schema analytics \
  --conn-login etl_user \
  --conn-password 's3cr3t' \
  --conn-port 5432

Successfully added `conn_id`=postgres_analytics : postgres://etl_user:******@db.internal:5432/analytics

airflow connections list --conn-id postgres_analytics

id | conn_id             | conn_type | host        | schema    | login    | port
===+=====================+===========+=============+===========+==========+=====
5  | postgres_analytics  | postgres  | db.internal | analytics | etl_user | 5432

The password is not shown and is stored Fernet-encrypted. Test it before writing a DAG around it:

airflow connections test postgres_analytics

Connection successfully tested

From the environment instead

For containers, a connection can come from an environment variable — no database write, no setup step:

export AIRFLOW_CONN_POSTGRES_ANALYTICS='postgresql://etl_user:s3cr3t@db.internal:5432/analytics'

from airflow.hooks.base import BaseHook
conn = BaseHook.get_connection("postgres_analytics")
print(f"{conn.conn_type}://{conn.login}@{conn.host}:{conn.port}/{conn.schema}")

postgres://etl_user@db.internal:5432/analytics

The variable name is AIRFLOW_CONN_ plus the connection ID uppercased. Environment connections take precedence over stored ones, which makes overriding a connection per environment straightforward.

Using a hook

from datetime import datetime
from airflow.decorators import dag, task
from airflow.providers.postgres.hooks.postgres import PostgresHook


@dag(dag_id="hooks_demo", start_date=datetime(2026, 2, 1), schedule=None, catchup=False)
def pipeline():

    @task
    def row_count() -> int:
        hook = PostgresHook(postgres_conn_id="postgres_analytics")
        result = hook.get_first("SELECT COUNT(*) FROM orders WHERE order_date = %s",
                                parameters=("2026-02-14",))
        print(f"rows: {result[0]}")
        return result[0]

    @task
    def load_summary(count: int) -> None:
        hook = PostgresHook(postgres_conn_id="postgres_analytics")
        hook.run(
            "INSERT INTO daily_summary (day, row_count) VALUES (%s, %s) "
            "ON CONFLICT (day) DO UPDATE SET row_count = EXCLUDED.row_count",
            parameters=("2026-02-14", count),
        )
        print(f"upserted summary: {count}")

    load_summary(row_count())


pipeline()

INFO - rows: 1284
INFO - upserted summary: 1284

Note the %s parameters rather than an f-string. Hooks pass parameters to the driver, which handles quoting — building SQL with string interpolation is a SQL injection waiting to happen even inside a “trusted” pipeline.

The ON CONFLICT DO UPDATE matters too: Airflow retries tasks, so every write should be safe to run twice.

Useful hook methods

    @task
    def hook_methods() -> None:
        hook = PostgresHook(postgres_conn_id="postgres_analytics")

        print("get_first:  ", hook.get_first("SELECT COUNT(*) FROM orders"))
        print("get_records:", hook.get_records("SELECT country, COUNT(*) FROM orders GROUP BY country LIMIT 3"))

        df = hook.get_pandas_df("SELECT country, SUM(amount) AS rev FROM orders GROUP BY country")
        print(df.to_string(index=False))

get_first:   (1284,)
get_records: [('UK', 512), ('US', 431), ('DE', 341)]
country     rev
     UK 128400.5
     US 143201.2
     DE  89042.7

get_pandas_df loads the whole result into the worker’s memory — fine for a summary, fatal for a full table.

S3

from airflow.providers.amazon.aws.hooks.s3 import S3Hook

    @task
    def list_and_load() -> list[str]:
        hook = S3Hook(aws_conn_id="aws_default")

        keys = hook.list_keys(bucket_name="data-lake", prefix="raw/2026-02-14/")
        print(f"found {len(keys)} objects")

        for k in keys[:3]:
            print(f"  {k}  ({hook.head_object(k, 'data-lake')['ContentLength']} bytes)")

        return keys

INFO - found 12 objects
INFO -   raw/2026-02-14/part-000.csv  (2481042 bytes)
INFO -   raw/2026-02-14/part-001.csv  (2390118 bytes)
INFO -   raw/2026-02-14/part-002.csv  (2447891 bytes)

Variables

airflow variables set batch_size 5000
airflow variables set slack_channel '#data-alerts'
airflow variables set feature_flags '{"use_new_parser": true, "dry_run": false}' --json

Variable batch_size created
Variable slack_channel created
Variable feature_flags created

from airflow.models import Variable

    @task
    def use_variables() -> None:
        size = int(Variable.get("batch_size"))
        flags = Variable.get("feature_flags", deserialize_json=True)
        missing = Variable.get("not_set", default_var="fallback")

        print(f"batch_size:  {size}")
        print(f"flags:       {flags}")
        print(f"missing:     {missing}")

batch_size:  5000
flags:       {'use_new_parser': True, 'dry_run': False}
missing:     fallback

The parse-time trap

This is the mistake worth measuring:

from airflow.models import Variable

BATCH_SIZE = int(Variable.get("batch_size"))       # module level — runs on every parse

with DAG(dag_id="bad_variables", ...) as dag:
    ...

The scheduler re-parses every DAG file on min_file_process_interval — 30 seconds by default. A module-level Variable.get() therefore issues a database query every 30 seconds, per DAG file, forever.

$ grep -c "SELECT variable" scheduler.log
2874

With 50 DAG files each reading two variables that is 200 queries a minute doing nothing. On a busy instance it visibly slows DAG parsing.

Read inside the task instead:

    @task
    def process() -> None:
        size = int(Variable.get("batch_size"))     # runs when the task runs
        print(f"batch size: {size}")

Or use templating, which the scheduler resolves lazily:

    load = BashOperator(
        task_id="load",
        bash_command="load.sh --batch {{ var.value.batch_size }} --day {{ ds }}",
    )

INFO - Running command: ['/bin/bash', '-c', 'load.sh --batch 5000 --day 2026-02-14']

Same rule applies to Connection.get and any other database access at module level: a DAG file should define structure, not fetch data.

Secrets backends

Keeping credentials out of Airflow’s database entirely:

# airflow.cfg
[secrets]
backend = airflow.providers.amazon.aws.secrets.secrets_manager.SecretsManagerBackend
backend_kwargs = {"connections_prefix": "airflow/connections", "variables_prefix": "airflow/variables"}

aws secretsmanager create-secret \
  --name airflow/connections/postgres_analytics \
  --secret-string 'postgresql://etl_user:s3cr3t@db.internal:5432/analytics'

{
    "ARN": "arn:aws:secretsmanager:eu-west-1:123456789012:secret:airflow/connections/postgres_analytics-Ab3xYz",
    "Name": "airflow/connections/postgres_analytics"
}

No DAG change is needed — PostgresHook(postgres_conn_id="postgres_analytics") now resolves from Secrets Manager. Airflow checks the secrets backend first, then environment variables, then its own database.

Masking

    @task
    def leaky() -> None:
        conn = BaseHook.get_connection("postgres_analytics")
        print(f"connecting with password {conn.password}")

INFO - connecting with password ***

Airflow masks known secret values in task logs automatically. Do not rely on it — it only knows values that came from connections and variables, so a password read from a file or built by string concatenation will appear in full.

Practice

1. Create a connection via environment variable and confirm it overrides the stored one.

export AIRFLOW_CONN_POSTGRES_ANALYTICS='postgresql://other:pw@other-host:5432/other'

postgres://other@other-host:5432/other

The environment wins. This is how the same DAG points at staging and production without a code change.

2. Put Variable.get() at module level and count scheduler queries.

module level: 2874 queries in one hour
inside task:     6 queries in one hour

Nearly 500× the load, for a value that only mattered six times. Any Variable.get() outside a task function or template is worth removing.

3. Store a JSON variable and read one key.

Variable.get("feature_flags", deserialize_json=True)["use_new_parser"]

True

One variable holding a config object beats a dozen scalar variables — one database read, and the values change together. In templates: {{ var.json.feature_flags.use_new_parser }}.

4. Print a connection password in a task log.

INFO - connecting with password ***

Masked, because the value came from a connection Airflow knows about. Build the same string from a file read and it prints in full — masking is a safety net, not a guarantee.

Next: testing a DAG before it reaches the scheduler.