Connection schema is returned as None when fetching via jinja template

[atul-astronomer]

Apache Airflow version

Tag: 3.0.0a1

What happened?

Connection schema is returned as None when fetching using f"{{{{ conn.{dag_name}_connection.schema }}}}"

Stack trace:

dag_id=test_jinja_connection_id_run_id=manual__2025-02-04T06_33_59.506710+00_00_task_id=check_jinja_conn_id_attempt=1.log

What you think should happen instead?

Connection schema value should be returned correct and assertion should pass.

How to reproduce

Use the below DAG to reproduce:
The connection can be created using UI or API having below parameters
request_body = { "connection_id": f"{dag_name}_connection", "conn_type": "postgres", "description": "postgres", "host": "database-1.cxmxicvi57az.us-east-2.rds.amazonaws.com", "login": "postgres", "schema": "postgres", "port": 5432, "password": POSTGRES_PASS }

import os

from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from airflow.providers.standard.operators.python import PythonOperator
from pendulum import today

from airflow.exceptions import AirflowNotFoundException
from airflow.hooks.base import BaseHook
from airflow.models import DAG
from dags.plugins import api_utility

dag_name = "test_jinja_connection_id"

def conn_id_test(**context):
    print("CONTEXT: ", context)
    print(f"The connection type is: {context['get_conn_type']}")
    print(f"The host is: {context['check_host']}")
    print(f"The schema is: {context['get_schema']}")
    print(f"The login is: {context['get_login']}")
    print(f"The password is: {context['get_pass']}")  # returns '***' hides the password
    print(f"The port is: {context['get_port']}")
    print(f"The extras are: {context['get_extras']}")

    print("asserting the connection type")
    assert context["get_conn_type"] == "postgres"
    print("asserting the host")
    assert context["check_host"] == POSTGRES_HOST
    print("asserting the schema")
    assert context["get_schema"] == "postgres"
    print("asserting the login")
    assert context["get_login"] == "postgres"
    print("asserting the port")
    assert context["get_port"] == "5432"
    # print("asserting the extras")
    # assert context["get_extras"] == "{'key': 'value'}"


with DAG(
    dag_id=dag_name,
    schedule=None,
    start_date=today('UTC').add(days=-2),
    doc_md=docs,
    tags=["core"],
) as dag:

    P0 = SQLExecuteQueryOperator(
        task_id="create_table_define_cols",
        conn_id=f"{dag_name}_connection",
        sql="""
            CREATE TABLE IF NOT EXISTS jinja_connection_template_test(
            random_str varchar,
            herbs varchar,
            primary key(herbs));
            """,
    )

    py1 = PythonOperator(
        task_id="check_jinja_conn_id",
        python_callable=conn_id_test,
        op_kwargs={
            "get_conn_type": f"{{{{ conn.{dag_name}_connection.conn_type }}}}",
            "check_host": f"{{{{ conn.{dag_name}_connection.host }}}}",
            "get_schema": f"{{{{ conn.{dag_name}_connection.schema }}}}",
            "get_login": f"{{{{ conn.{dag_name}_connection.login }}}}",
            "get_pass": f"{{{{ conn.{dag_name}_connection.password }}}}",
            "get_port": f"{{{{ conn.{dag_name}_connection.port }}}}",
            "get_extras": f"{{{{ conn.{dag_name}_connection.extra }}}}",
        },
    )


P0 >> py1

Operating System

Linux

Versions of Apache Airflow Providers

No response

Deployment

Other

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct