Local Filesystem secret backend unable to retrieve connection

[adrian-edbert]

Apache Airflow version

3.1.0

If "Other Airflow 2/3 version" selected, which one?

No response

What happened?

Airflow worker failed on getting local filesystem secret backend
With an error message: unable to retrieve connection

What you think should happen instead?

Airflow worker should retrieve connection coming from local filesystem secrets backend

How to reproduce

with dag file

from __future__ import annotations

from airflow.providers.standard.operators.python import PythonOperator

def print_connection():
    from airflow.sdk.bases.hook import BaseHook
    worker_conn = BaseHook.get_connection("worker_connection_env")
    print(f"worker_connection_env : {worker_conn}")

with DAG(
    dag_id='test_dag',
    catchup=False,
):
    test = PythonOperator(task_id='task3', python_callable=print_connection)
    test

on a worker configuration with this env var

AIRFLOW__SECRETS__BACKEND=airflow.secrets.local_filesystem.LocalFilesystemBackend
AIRFLOW__SECRETS__BACKEND_KWARGS={"connections_file_path": "{path_to_config}/connection.yaml"}

connection.yaml

worker_connection_env:
  conn_type: Generic
  description: worker specific connection env for sanity test

this will fail with these message

[2025-10-22 11:31:52] ERROR - Unable to retrieve connection from secrets backend (LocalFilesystemBackend). Checking subsequent secrets backend. source=task loc=context.py:162
InvalidRequestError: When initializing mapper mapped class DagVersion->dag_version, expression 'DagCode' failed to locate a name ('DagCode'). If this is a class name, consider adding this relationship() to the <class 'airflow.models.dag_version.DagVersion'> class after both dependent classes have been defined.
File "/home/generic_scheduler/test_airflow/generic_scheduler-venv/lib/python3.12/site-packages/airflow/sdk/execution_time/context.py", line 156 in _get_connection

File "/home/generic_scheduler/test_airflow/generic_scheduler-venv/lib/python3.12/site-packages/airflow/secrets/local_filesystem.py", line 314 in get_connection

File "/home/generic_scheduler/test_airflow/generic_scheduler-venv/lib/python3.12/site-packages/airflow/secrets/local_filesystem.py", line 311 in _local_connections

File "/home/generic_scheduler/test_airflow/generic_scheduler-venv/lib/python3.12/site-packages/airflow/secrets/local_filesystem.py", line 273 in load_connections_dict

File "/home/generic_scheduler/test_airflow/generic_scheduler-venv/lib/python3.12/site-packages/airflow/secrets/local_filesystem.py", line 228 in _create_connection

File "<string>", line 4 in __init__

File "/home/generic_scheduler/test_airflow/generic_scheduler-venv/lib/python3.12/site-packages/sqlalchemy/orm/state.py", line 476 in _initialize_instance

File "/home/generic_scheduler/test_airflow/generic_scheduler-venv/lib/python3.12/site-packages/sqlalchemy/event/attr.py", line 346 in __call__

File "/home/generic_scheduler/test_airflow/generic_scheduler-venv/lib/python3.12/site-packages/sqlalchemy/orm/mapper.py", line 3716 in _event_on_init

File "/home/generic_scheduler/test_airflow/generic_scheduler-venv/lib/python3.12/site-packages/sqlalchemy/orm/mapper.py", line 1941 in _check_configure

File "/home/generic_scheduler/test_airflow/generic_scheduler-venv/lib/python3.12/site-packages/sqlalchemy/orm/mapper.py", line 3527 in _configure_registries

File "/home/generic_scheduler/test_airflow/generic_scheduler-venv/lib/python3.12/site-packages/sqlalchemy/orm/mapper.py", line 3566 in _do_configure_registries

File "/home/generic_scheduler/test_airflow/generic_scheduler-venv/lib/python3.12/site-packages/sqlalchemy/orm/mapper.py", line 1958 in _post_configure_properties

File "/home/generic_scheduler/test_airflow/generic_scheduler-venv/lib/python3.12/site-packages/sqlalchemy/orm/interfaces.py", line 231 in init

File "/home/generic_scheduler/test_airflow/generic_scheduler-venv/lib/python3.12/site-packages/sqlalchemy/orm/relationships.py", line 2150 in do_init

File "/home/generic_scheduler/test_airflow/generic_scheduler-venv/lib/python3.12/site-packages/sqlalchemy/orm/relationships.py", line 2245 in _process_dependent_arguments

File "/home/generic_scheduler/test_airflow/generic_scheduler-venv/lib/python3.12/site-packages/sqlalchemy/util/langhelpers.py", line 1113 in __get__

File "/home/generic_scheduler/test_airflow/generic_scheduler-venv/lib/python3.12/site-packages/sqlalchemy/orm/relationships.py", line 2112 in entity

File "/home/generic_scheduler/test_airflow/generic_scheduler-venv/lib/python3.12/site-packages/sqlalchemy/orm/clsregistry.py", line 397 in _resolve_name

File "/home/generic_scheduler/test_airflow/generic_scheduler-venv/lib/python3.12/site-packages/sqlalchemy/orm/clsregistry.py", line 375 in _raise_for_name

File "/home/generic_scheduler/test_airflow/generic_scheduler-venv/lib/python3.12/site-packages/sqlalchemy/util/compat.py", line 211 in raise_

KeyError: 'DagCode'
File "/home/generic_scheduler/test_airflow/generic_scheduler-venv/lib/python3.12/site-packages/sqlalchemy/orm/clsregistry.py", line 393 in _resolve_name

File "/home/generic_scheduler/test_airflow/generic_scheduler-venv/lib/python3.12/site-packages/sqlalchemy/util/_collections.py", line 746 in __missing__

File "/home/generic_scheduler/test_airflow/generic_scheduler-venv/lib/python3.12/site-packages/sqlalchemy/orm/clsregistry.py", line 372 in _access_cls

Operating System

Ubuntu 20.04.5 LTS (Focal Fossa)

Versions of Apache Airflow Providers

No response

Deployment

Virtualenv installation

Deployment details

Virtual env installation using UV
pip freeze | grep -i airflow

apache-airflow==3.1.0
apache-airflow-client==3.1.0rc1
apache-airflow-core==3.1.0.1
apache-airflow-providers-apache-hdfs==4.10.1
apache-airflow-providers-apache-spark==5.3.1
apache-airflow-providers-celery==3.12.1
apache-airflow-providers-common-compat==1.7.2
apache-airflow-providers-common-io==1.6.1
apache-airflow-providers-common-sql==1.27.3
apache-airflow-providers-fab==2.4.4
apache-airflow-providers-git==0.0.8
apache-airflow-providers-http==5.4.0rc1
apache-airflow-providers-smtp==2.1.1
apache-airflow-providers-standard==1.4.1
apache-airflow-task-sdk==1.1.0

Anything else?

From our test on our fork, changing the import
from airflow.models.connection import Connection
into
from airflow.sdk.definitions.connection import Connection

in file airflow-core/src/airflow/secrets/local_filesystem.py

works to fix the issue

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

[amoghrajesh]

The local file system backend uses connection model but it has a back compat shim that should take care of the import of sdk vs models. I tried with the same dag and configuration as you @adrian-edbert and it works just fine for me.

DAG:

from __future__ import annotations

from airflow.providers.standard.operators.python import PythonOperator
from airflow import DAG


def print_connection():
    from airflow.sdk.bases.hook import BaseHook
    worker_conn = BaseHook.get_connection("worker_connection_env")
    print(f"worker_connection_env : {worker_conn}")


with DAG(
    dag_id='test_dag',
    catchup=False,
    schedule=None
) as dag:
    test = PythonOperator(task_id='task3', python_callable=print_connection)
    test

Could you explain what you have done different?

[adrian-edbert]

The local file system backend uses connection model but it has a back compat shim that should take care of the import of sdk vs models. I tried with the same dag and configuration as you @adrian-edbert and it works just fine for me.

DAG:

from future import annotations

from airflow.providers.standard.operators.python import PythonOperator
from airflow import DAG

def print_connection():
from airflow.sdk.bases.hook import BaseHook
worker_conn = BaseHook.get_connection("worker_connection_env")
print(f"worker_connection_env : {worker_conn}")

with DAG(
dag_id='test_dag',
catchup=False,
schedule=None
) as dag:
test = PythonOperator(task_id='task3', python_callable=print_connection)
test
Could you explain what you have done different?

oh I couldn't find the compat shim for this one, mind pointing me to the compat definition file?
I do have some logic changes in my fork for the git dag bundle, but I don't think that one should affect this? I will double check without git dag bundle and get back to this issue again

[adrian-edbert]

Tested some more, previously I am using our own fork for the worker, reinstalled using the pypi package directly

I still can't get it to work, my exact setup for the worker is using this uv script to install

# pwd = /home/generic_scheduler/test_airflow/
uv venv --seed --python 3.12 test-venv
uv pip install apache-airflow[celery]==3.1.0 --python test-venv --constraint https://raw.githubusercontent.com/apache/airflow/constraints-3.1.0/constraints-3.12.txt
uv pip install apache-airflow[celery]==3.1.0 --python test-venv -r test_airflow/generic_scheduler-venv/worker_requirements.txt 

worker_requirements.txt

apache-airflow-client
apache-airflow[statsd]==3.1.0
mysqlclient==2.2.7
pymysql==1.1.1
aiomysql==0.2.0
authlib==1.6.0
apache-airflow-providers-fab==2.4.4
apache-airflow-providers-git==0.0.8
apache-airflow-providers-apache-spark==5.3.1
apache-airflow-providers-apache-hdfs==4.10.1

this is run using systemd

[Unit]
Description=Airflow celery worker daemon
After=network.target

[Service]
EnvironmentFile=/home/generic_scheduler/test_airflow/generic_scheduler.systemd.env
User=generic_scheduler
Type=simple
ExecStart=/bin/bash -c "/home/generic_scheduler/test_airflow/test-venv/bin/airflow celery worker -q test_queue --pid /home/generic_scheduler/test_airflow/run/worker.pid"
Restart=on-failure
RestartSec=10s

[Install]
WantedBy=multi-user.target

generic_scheduler.systemd.env

AIRFLOW_HOME=/home/generic_scheduler/test_airflow
AIRFLOW__SECRETS__BACKEND=airflow.secrets.local_filesystem.LocalFilesystemBackend
AIRFLOW__SECRETS__BACKEND_KWARGS={"connections_file_path": "/home/generic_scheduler/test_airflow/connection.yaml"}

connection.yaml

worker_connection_env:
  conn_type: Generic
  description: worker specific connection env for sanity test

the full dag looks like this, I omitted some parts before because I don't think it's relevant (the queue and access_control), the operator still the same

with DAG(
    dag_id='testdag',
    catchup=False,
    default_args = {
        "queue": "test_queue",
        "owner": "user",
        "run_as_user": "user"
    },
    access_control={'test_role':{'DAGs': {'can_read'}}, 'test_role_ops':{"DAGs": {"can_edit"}, "DAG Runs": {"can_create"},}},
    tags=["sanity_test"]
):

I am still not sure why the compat shim didn't work for my setup

[amoghrajesh]

Looking at your stacktrace, I see this error:

When the flow reached this

local_filesystem.py:228 in _create_connection

This starts:

<string>:4 in __init__
↓
sqlalchemy/orm/state.py:476 in _initialize_instance
↓
sqlalchemy/event/attr.py:346 in __call__
↓
sqlalchemy/orm/mapper.py:3716 in _event_on_init
↓
sqlalchemy/orm/mapper.py:1941 in _check_configure

sqlalchemy/orm/mapper.py:3527 in _configure_registries
↓
sqlalchemy/orm/mapper.py:3566 in _do_configure_registries

sqlalchemy/orm/mapper.py:1958 in _post_configure_properties
↓
sqlalchemy/orm/relationships.py:2150 in do_init
↓
sqlalchemy/orm/relationships.py:2245 in _process_dependent_arguments

sqlalchemy/orm/clsregistry.py:397 in _resolve_name
↓
KeyError: 'DagCode'

The case seems to be that the worker environment is incomplete. My hypothesis: In your Celery worker, some initialization code (in your fork) imports DagVersion without importing DagCode, causing the mapper configuration to fail.

@adrian-edbert TLDR, could you check what happens outside your fork so you can rule out if your changes cause this?

[adrian-edbert]

the last configuration I pass is using the pypi package without involving my fork

I have also tried using the newly published 3.1.1 pypi package, but now I got different error which basically just mention the connection can't be found

[2025-10-28 14:31:37] ERROR - Task failed with exception source=task loc=task_runner.py:972
AirflowNotFoundException: The conn_id `worker_connection_env` isn't defined
File "/home/generic_scheduler/test_airflow/test-venv/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 920 in run

File "/home/generic_scheduler/test_airflow/test-venv/lib/python3.12/site-packages/airflow/sdk/execution_time/task_runner.py", line 1307 in _execute_task

File "/home/generic_scheduler/test_airflow/test-venv/lib/python3.12/site-packages/airflow/sdk/bases/operator.py", line 416 in wrapper

File "/home/generic_scheduler/test_airflow/test-venv/lib/python3.12/site-packages/airflow/providers/standard/operators/python.py", line 216 in execute

File "/home/generic_scheduler/test_airflow/test-venv/lib/python3.12/site-packages/airflow/providers/standard/operators/python.py", line 239 in execute_callable

File "/home/generic_scheduler/test_airflow/test-venv/lib/python3.12/site-packages/airflow/sdk/execution_time/callback_runner.py", line 82 in run

File "/home/generic_scheduler/test_airflow/dag-bundles/...../versions/...../dags/test_dag.py", line 25 in print_env

File "/home/generic_scheduler/test_airflow/test-venv/lib/python3.12/site-packages/airflow/sdk/bases/hook.py", line 61 in get_connection

File "/home/generic_scheduler/test_airflow/test-venv/lib/python3.12/site-packages/airflow/sdk/definitions/connection.py", line 225 in get

File "/home/generic_scheduler/test_airflow/test-venv/lib/python3.12/site-packages/airflow/sdk/execution_time/context.py", line 172 in _get_connection

for now since it seems to be working from your side, I'll try to check if there's any changes from my fork that initialize DagVersion or something similar along the celery worker, that seems to be the things that differ in my envs

[amoghrajesh]

Sounds good, it seems that the connection is indeed not present. The connection.yaml should be present on workers too btw. Check for that as well.

IAC, are we OK to close this issue?

[adrian-edbert]

yes, I check connection.yaml is already present on the worker
still not sure why my setup doesn't work, my suspicion is because I am using the run_as_user and supervisor have different behavior when getting connection inside the worker, but I don't have spare time to check on it

can close the issue for now since it seems not reproducible, thank you for the help