amoghrajesh opened a new pull request, #52968:
URL: https://github.com/apache/airflow/pull/52968
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
closes: (https://github.com/apache/airflow/issues/52880)
## Problem
Accessing a connection from SDK (Connection or using context) in airflow 3
and if connection isn't found it raises: `AirflowRuntimeError:
CONNECTION_NOT_FOUND: {'conn_id': 'my_conn'}`
This is the case when pure SDK is used: `airflow.sdk.Connection`
(airflow.models will work fine as we wrap it well)
Airflow 2 raises: `airflow.exceptions.AirflowNotFoundException: The conn_id
`my_conn` isn't defined` .
Now due to this, some dags / providers use this for various sorts of
filtering and for catching exceptions to perform some logic. So long term, this
current way of handling exceptions is not feasible. We will start seeing
problems when we start migrating providers to Airflow 3 entirely. Example
issues: https://github.com/apache/airflow/pull/52838 by amazon team and also
https://github.com/apache/airflow/issues/52921 by bosch team.
## Testing the solution
DAG used:
```
from __future__ import annotations
from airflow.models.dag import dag
from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import Connection
def get_conn_context(**context):
conn = context["conn"].my_conn
return conn
def get_conn_sdk(**context):
return Connection.get("my_conn")
def get_conn_models(**context):
from airflow.models import Connection
return Connection.get_connection_from_secrets("my_conn")
@dag()
def get_connection():
a = PythonOperator(
task_id="context_conn",
python_callable=get_conn_context,
)
b = PythonOperator(
task_id="sdk_conn",
python_callable=get_conn_sdk,
)
c = PythonOperator(
task_id="models_conn",
python_callable=get_conn_models,
)
[a, b, c]
get_connection()
```
Without the fix:
Accessing from context(task a):
```
[2025-07-07, 13:41:04] INFO - DAG bundles loaded: dags-folder:
source="airflow.dag_processing.bundles.manager.DagBundlesManager"
[2025-07-07, 13:41:04] INFO - Filling up the DagBag from
/files/dags/get-connection.py: source="airflow.models.dagbag.DagBag"
[2025-07-07, 13:41:04] ERROR - Task failed with exception: source="task"
AirflowRuntimeError: CONNECTION_NOT_FOUND: {'conn_id': 'my_conn'}
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py",
line 875 in run
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py",
line 1162 in _execute_task
File "/opt/airflow/task-sdk/src/airflow/sdk/bases/operator.py", line 397 in
wrapper
File
"/opt/airflow/providers/standard/src/airflow/providers/standard/operators/python.py",
line 217 in execute
File
"/opt/airflow/providers/standard/src/airflow/providers/standard/operators/python.py",
line 240 in execute_callable
File
"/opt/airflow/task-sdk/src/airflow/sdk/execution_time/callback_runner.py", line
82 in run
File "/files/dags/get-connection.py", line 9 in get_conn_context
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/context.py", line
275 in __getattr__
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/context.py", line
158 in _get_connection
```
Accessing from Models (task c):
```
[2025-07-07, 13:41:04] INFO - DAG bundles loaded: dags-folder:
source="airflow.dag_processing.bundles.manager.DagBundlesManager"
[2025-07-07, 13:41:04] INFO - Filling up the DagBag from
/files/dags/get-connection.py: source="airflow.models.dagbag.DagBag"
[2025-07-07, 13:41:04] ERROR - Task failed with exception: source="task"
AirflowNotFoundException: The conn_id `my_conn` isn't defined
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py",
line 875 in run
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py",
line 1162 in _execute_task
File "/opt/airflow/task-sdk/src/airflow/sdk/bases/operator.py", line 397 in
wrapper
File
"/opt/airflow/providers/standard/src/airflow/providers/standard/operators/python.py",
line 217 in execute
File
"/opt/airflow/providers/standard/src/airflow/providers/standard/operators/python.py",
line 240 in execute_callable
File
"/opt/airflow/task-sdk/src/airflow/sdk/execution_time/callback_runner.py", line
82 in run
File "/files/dags/get-connection.py", line 18 in get_conn_models
File "/opt/airflow/airflow-core/src/airflow/models/connection.py", line 490
in get_connection_from_secrets
AirflowRuntimeError: CONNECTION_NOT_FOUND: {'conn_id': 'my_conn'}
File "/opt/airflow/airflow-core/src/airflow/models/connection.py", line 481
in get_connection_from_secrets
File "/opt/airflow/task-sdk/src/airflow/sdk/definitions/connection.py", line
152 in get
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/context.py", line
158 in _get_connection
```
Accessing from SDK:
```
[2025-07-07, 13:41:04] INFO - DAG bundles loaded: dags-folder:
source="airflow.dag_processing.bundles.manager.DagBundlesManager"
[2025-07-07, 13:41:04] INFO - Filling up the DagBag from
/files/dags/get-connection.py: source="airflow.models.dagbag.DagBag"
[2025-07-07, 13:41:04] ERROR - Task failed with exception: source="task"
AirflowRuntimeError: CONNECTION_NOT_FOUND: {'conn_id': 'my_conn'}
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py",
line 875 in run
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py",
line 1162 in _execute_task
File "/opt/airflow/task-sdk/src/airflow/sdk/bases/operator.py", line 397 in
wrapper
File
"/opt/airflow/providers/standard/src/airflow/providers/standard/operators/python.py",
line 217 in execute
File
"/opt/airflow/providers/standard/src/airflow/providers/standard/operators/python.py",
line 240 in execute_callable
File
"/opt/airflow/task-sdk/src/airflow/sdk/execution_time/callback_runner.py", line
82 in run
File "/files/dags/get-connection.py", line 13 in get_conn_sdk
File "/opt/airflow/task-sdk/src/airflow/sdk/definitions/connection.py", line
152 in get
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/context.py", line
158 in _get_connection
```
After changes:
Accessing from context (task a):
```
[2025-07-07, 13:51:33] INFO - DAG bundles loaded: dags-folder:
source="airflow.dag_processing.bundles.manager.DagBundlesManager"
[2025-07-07, 13:51:33] INFO - Filling up the DagBag from
/files/dags/get-connection.py: source="airflow.models.dagbag.DagBag"
[2025-07-07, 13:51:33] ERROR - Task failed with exception: source="task"
AirflowNotFoundException: The conn_id `my_conn` isn't defined
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py",
line 875 in run
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py",
line 1162 in _execute_task
File "/opt/airflow/task-sdk/src/airflow/sdk/bases/operator.py", line 397 in
wrapper
File
"/opt/airflow/providers/standard/src/airflow/providers/standard/operators/python.py",
line 217 in execute
File
"/opt/airflow/providers/standard/src/airflow/providers/standard/operators/python.py",
line 240 in execute_callable
File
"/opt/airflow/task-sdk/src/airflow/sdk/execution_time/callback_runner.py", line
82 in run
File "/files/dags/get-connection.py", line 9 in get_conn_context
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/context.py", line
280 in __getattr__
AirflowRuntimeError: CONNECTION_NOT_FOUND: {'conn_id': 'my_conn'}
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/context.py", line
277 in __getattr__
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/context.py", line
159 in _get_connection
```
Accessing from models (task b):
```
[2025-07-07, 13:51:33] INFO - DAG bundles loaded: dags-folder:
source="airflow.dag_processing.bundles.manager.DagBundlesManager"
[2025-07-07, 13:51:33] INFO - Filling up the DagBag from
/files/dags/get-connection.py: source="airflow.models.dagbag.DagBag"
[2025-07-07, 13:51:33] ERROR - Task failed with exception: source="task"
AirflowNotFoundException: The conn_id `my_conn` isn't defined
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py",
line 875 in run
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py",
line 1162 in _execute_task
File "/opt/airflow/task-sdk/src/airflow/sdk/bases/operator.py", line 397 in
wrapper
File
"/opt/airflow/providers/standard/src/airflow/providers/standard/operators/python.py",
line 217 in execute
File
"/opt/airflow/providers/standard/src/airflow/providers/standard/operators/python.py",
line 240 in execute_callable
File
"/opt/airflow/task-sdk/src/airflow/sdk/execution_time/callback_runner.py", line
82 in run
File "/files/dags/get-connection.py", line 18 in get_conn_models
File "/opt/airflow/airflow-core/src/airflow/models/connection.py", line 481
in get_connection_from_secrets
File "/opt/airflow/task-sdk/src/airflow/sdk/definitions/connection.py", line
156 in get
AirflowRuntimeError: CONNECTION_NOT_FOUND: {'conn_id': 'my_conn'}
File "/opt/airflow/task-sdk/src/airflow/sdk/definitions/connection.py", line
153 in get
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/context.py", line
159 in _get_connection
```
Accessing from SDK (task c):
```
[2025-07-07, 13:51:33] INFO - DAG bundles loaded: dags-folder:
source="airflow.dag_processing.bundles.manager.DagBundlesManager"
[2025-07-07, 13:51:33] INFO - Filling up the DagBag from
/files/dags/get-connection.py: source="airflow.models.dagbag.DagBag"
[2025-07-07, 13:51:33] ERROR - Task failed with exception: source="task"
AirflowNotFoundException: The conn_id `my_conn` isn't defined
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py",
line 875 in run
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py",
line 1162 in _execute_task
File "/opt/airflow/task-sdk/src/airflow/sdk/bases/operator.py", line 397 in
wrapper
File
"/opt/airflow/providers/standard/src/airflow/providers/standard/operators/python.py",
line 217 in execute
File
"/opt/airflow/providers/standard/src/airflow/providers/standard/operators/python.py",
line 240 in execute_callable
File
"/opt/airflow/task-sdk/src/airflow/sdk/execution_time/callback_runner.py", line
82 in run
File "/files/dags/get-connection.py", line 13 in get_conn_sdk
File "/opt/airflow/task-sdk/src/airflow/sdk/definitions/connection.py", line
156 in get
AirflowRuntimeError: CONNECTION_NOT_FOUND: {'conn_id': 'my_conn'}
File "/opt/airflow/task-sdk/src/airflow/sdk/definitions/connection.py", line
153 in get
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/context.py", line
159 in _get_connection
```
<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request
Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)**
for more information.
In case of fundamental code changes, an Airflow Improvement Proposal
([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals))
is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party
License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a
newsfragment file, named `{pr_number}.significant.rst` or
`{issue_number}.significant.rst`, in
[airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
