Airflow Connections → Orchestra Connections

Overview

Airflow never puts credentials directly in DAG code. Instead, every operator or hook takes a conn_id (or a provider-specific variant like snowflake_conn_id, aws_conn_id, ssh_conn_id, slack_conn_id, gcp_conn_id) that is a pointer to a Connection record — stored in the metadata DB, an AIRFLOW_CONN_* env var, or an external Secrets Backend. The DAG source almost never contains the real host/user/password; it only contains the conn_id string. This skill teaches how to recognize each of Airflow's idioms for expressing that pointer and map it to an Orchestra connection — including the full connection-type table, naming rules, and secrets handling in one place, the way dagster-connections-to-orchestra and prefect-connections-to-orchestra do for their own orchestrators.

Connection Name Format

connection: my_snowflake_12345   # format: descriptive-name_XXXXX (5-digit suffix from UI)

The 5-digit suffix is assigned by Orchestra when the connection is created — never invent it. Read it back rather than asking someone to find it: connections are managed through the public API and the Orchestra MCP server as well as the UI, so listing the workspace's connections returns each one's descriptive_name_NNNNN id directly. Fall back to copying it from the UI only where neither is available.

No conn_id referenced? If the task doesn't actually take a _conn_id=/conn_id= kwarg or call BaseHook.get_connection(...) — pure computation, no external client, no secrets — don't invent a connection name or a fake env var placeholder just to fill the field:

connection: null   # no distinct conn_id in the source; Orchestra uses the workspace default for this integration

This extends to task parameters that duplicate connection-level scope, too — e.g. Power BI's workspace_id (from group_id), or any other parameter whose value is also stored on the Orchestra connection itself. If the source code just reads the same single value everywhere rather than genuinely varying it per task, leave that parameter null/omitted and let the connection's own configured value apply.

For environment-specific connections:

connection: ${{ ENV.SNOWFLAKE_CONNECTION_NAME }}

Set SNOWFLAKE_CONNECTION_NAME=my_snowflake_12345 in Orchestra's environment settings.

Airflow Credential Idioms → What They Mean

In every one of these cases, the Orchestra-side answer is the same: create one Orchestra connection per distinct conn_id and reference it by name in the YAML connection: field. The only thing that differs is how much information you can recover from source to help a human set up that connection correctly.

conn_id kwarg → Orchestra Connection Type

When conn_id= is generic (no provider-specific kwarg name), the operator class itself tells you the integration — e.g. conn_id="powerbi_default" on a Power BI refresh operator still maps to Orchestra's Power BI connection type, not a generic HTTP connection.

Databases

Data Integration

Notifications

Infrastructure & Protocol

Secrets Backend Gotcha

If the Airflow deployment configures a Secrets Backend (AWS Secrets Manager, Hashicorp Vault, GCP Secret Manager) for connections, the DAG code you're converting will show only the conn_id — there is no metadata-DB row, no AIRFLOW_CONN_* env var, nothing else in source to extract. This is expected and does not change the conversion:

• Still create one normal Orchestra connection for that conn_id.
• Note in your conversion output that the real credential values must be retrieved from the Secrets Backend (or from whoever administers it) rather than from anything visible in the DAG repo — there is nothing further to grep for.
• Never guess at a host/user/password to fill in a placeholder; leave the Orchestra connection fields for the user to complete from the actual secret.

If the backend is AWS Secrets Manager or Azure Key Vault, the credential may not need re-entering at all. An Orchestra connection can read its credentials straight from either of those stores instead of holding them itself: create an Orchestra connection to the store, then set Credential Store on the target connection and identify the secret — the Secret ARN for AWS Secrets Manager, the secret name for Azure Key Vault. Two constraints to flag when you propose this:

• A store only appears in the Credential Store dropdown once a connection to that store exists, so create the store connection first.
• The secret's value must be a JSON object holding exactly the keys the target integration expects — the parameter IDs from its Authentication parameters table, and every required one, not only the sensitive fields. An Airflow URI or extra blob is not that shape, so a secret written for Airflow usually has to be rewritten as the Orchestra connection's parameters before it can be read this way. In Azure Key Vault the whole JSON object is the secret's value, and the secret's name may contain only letters, numbers and hyphens — an Airflow conn_id with underscores can't be reused verbatim.
• Orchestra reads the secret when the connection is saved and rejects it if it can't be read or its keys don't match, so a mismatch surfaces at save time rather than mid-run. After that it reads the latest version on each use, so rotating the secret in the store needs no change in Orchestra.

Hashicorp Vault and GCP Secret Manager are not supported as Orchestra credential stores, so a DAG backed by either still means a manual re-entry. Both supported stores are in beta, so confirm they're enabled on the account before proposing this route. See External secret storage and Azure Key Vault.

Linked Connections — Reuse a Connection, Don't Copy Its Secrets

Not every conn_id is consumed by a dedicated operator: a PythonOperator that calls BaseHook.get_connection("snowflake_prod").password and builds its own client becomes an Orchestra Python (or dbt Core) task that still needs those warehouse credentials at runtime.

Decision rule: if the credential already exists as its own Orchestra connection — the Snowflake/Postgres/AWS connection you created from the tables above — link it on the Python/dbt Core connection (Linked connections on that connection's form) instead of re-typing the same user/password into that connection's Secrets. One credential, one place to rotate it.

Each linked connection's fields arrive as environment variables named <INTEGRATION>__<CONNECTION_ID>__<FIELD>: every part upper-cased, each run of non-alphanumeric characters collapsed to a single _, and the parts joined by a double underscore. <CONNECTION_ID> is the id the YAML's connection: field uses (it survives renames), and <FIELD> is the field's ID from that integration's Authentication parameters table — so a Snowflake connection prod_12345 exports SNOWFLAKE__PROD_12345__SNOWFLAKE_PASSWORD.

• Python reads them via the pre-installed SDK: orchestra.get_linked_connection("snowflake", connection_id="prod_12345"), keyed by field ID.
• dbt Core reads them in profiles.yml: {{ env_var('SNOWFLAKE__PROD_12345__SNOWFLAKE_PASSWORD') }}.
• A linked connection that has been deleted, or whose credentials can't be decrypted, is skipped silently — its variables are simply absent. Check a variable exists rather than assuming it.
• Linking is also what opens an approved private connection for a task on Orchestra-hosted compute (MariaDB today); private connections are not opened for connections that aren't linked.

Full reference: Linked connections (same section on dbt Core).

Variable / AIRFLOW_VAR_* vs Connection / AIRFLOW_CONN_*

These look similar (AIRFLOW_VAR_ vs AIRFLOW_CONN_ env var prefixes) but map to opposite places in Orchestra:

# Airflow — Variable is config, not a secret
target_table = Variable.get("target_table", default_var="orders")

# Airflow — Connection is a credential
hook = SnowflakeHook(snowflake_conn_id="snowflake_prod")

# Orchestra
inputs:
  target_table:
    type: string
    default: orders

pipeline:
  stage-load:
    tasks:
      load-data:
        integration: SNOWFLAKE
        integration_job: SNOWFLAKE_RUN_QUERY
        connection: snowflake_prod_12345   # from snowflake_conn_id="snowflake_prod"
        parameters:
          statement: 'INSERT INTO ${{ inputs.target_table }} SELECT * FROM staging'

If a Variable value happens to look secret-shaped (an API key someone stashed as a Variable instead of a Connection), flag it — Airflow Variables are not encrypted by default unless the key name matches a configured "sensitive" pattern, so treat it with the same caution as a Prefect Variable.get() (see prefect-secrets-to-orchestra for the parallel case in another orchestrator).

Conversion Checklist

• Grep the DAG source for every _conn_id= kwarg, BaseHook.get_connection(, and Connection( usage
• Grep the deployment/env files for AIRFLOW_CONN_* (connections) separately from AIRFLOW_VAR_* (config)
• For each distinct conn_id, identify the operator/hook it's passed to and look up the Orchestra connection type (table above)
• Check whether a Secrets Backend is configured (airflow.cfg [secrets] backend) — if so, note that no further credential detail can be extracted from source, and when that backend is AWS Secrets Manager or Azure Key Vault, propose pointing the Orchestra connection at the same secret via Credential Store rather than re-keying it
• For a conn_id consumed inside Python/dbt code rather than by a dedicated operator, link that credential's own Orchestra connection on the Python/dbt Core connection instead of copying its secrets across
• Create the Orchestra connection — in the UI (Settings → Connections), or through the public API / Orchestra MCP, which is worth scripting when a DAG estate has more than a handful of distinct conn_ids — and read back its name_XXXXX id from the response rather than transcribing it
• Replace every Airflow conn_id="..." reference with the matching Orchestra connection: name_XXXXX in the YAML
• Route Variable.get() / AIRFLOW_VAR_* values to inputs:, never to a connection
• Never write a real credential value (from an AIRFLOW_CONN_* URI or anywhere else) into the Orchestra YAML or any file — connections hold the secrets. Creating a connection programmatically doesn't change that: the values come from the user or their secret store at the moment of creation, never from a file you commit

Gotchas

• One conn_id can be reused across many tasks/DAGs — map it to a single Orchestra connection, don't create a duplicate per task.
• conn_id= is generic; the operator class disambiguates the type — the same generic kwarg name is used by Power BI, HTTP, and many community-provider operators, so identify the type from the operator/hook class, not the kwarg name alone.
• AIRFLOW_CONN_* URIs are real credentials in plain sight — if a deployment manifest or .env file has one, treat the value as sensitive; don't echo it back into any generated file, only use it to identify what fields the new Orchestra connection needs.
• Don't copy a warehouse credential into a Python/dbt connection's Secrets — if it already exists as its own Orchestra connection, link it and read the injected <INTEGRATION>__<CONNECTION_ID>__<FIELD> variables; a copied secret is a second copy to rotate, and a linked one that's deleted just goes missing rather than failing loudly.
• Secrets Backend means nothing to extract — this is normal, not a blocker; still create the Orchestra connection, just leave the value fields for the user. If that backend is AWS Secrets Manager or Azure Key Vault, offer the external credential store route instead of a manual re-key — the secret stays where it is and Orchestra reads it.
• Variable is not a secret store — even though it's fetched with a .get() call that superficially resembles Secret.load() in other orchestrators, Airflow Variables map to inputs:, not to a connection.
• dbt-over-SSH still isn't LINUX_SSH — if a conn_id/ssh_conn_id is feeding a dbt CLI invocation, defer to dbt-core-airflow-to-orchestra, not this skill's SSH mapping.

References

• Airflow connections concepts: https://airflow.apache.org/docs/apache-airflow/stable/authoring-and-scheduling/connections.html
• Airflow secrets backends: https://airflow.apache.org/docs/apache-airflow/stable/security/secrets/secrets-backend/index.html
• Orchestra connections: https://docs.getorchestra.io/docs/core-concepts/tasks/connections
• Managing connections programmatically (public API and MCP): https://docs.getorchestra.io/docs/core-concepts/tasks/connections#can-i-manage-connections-programmatically
• Orchestra external secret storage: https://docs.getorchestra.io/docs/core-concepts/tasks/connections#external-secret-storage
• Orchestra linked connections: https://docs.getorchestra.io/docs/integrations/python#linked-connections
• Orchestra environments: https://docs.getorchestra.io/docs/core-concepts/environments