Setting a variable in Dag is failing due to 'Direct database access via the ORM is not allowed in Airflow 3.0'

[atul-astronomer]

Apache Airflow version

3.0.0

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

No response

What happened?

Setting a variable in Dag is failing due to 'Direct database access via the ORM is not allowed in Airflow 3.0'

What you think should happen instead?

No response

How to reproduce

Have the below dag in dags folder, you will get import error.

from datetime import datetime

from airflow.providers.standard.operators.bash import BashOperator
from airflow.models import Variable

from airflow import DAG

my_var = Variable.set("param_variable", 10)

dag = DAG(
    'test_api_dag',
    start_date=datetime(2025, 3, 1, 3, 28, 0),
    # # schedule=timedelta(days=1),
    schedule='@daily',
    is_paused_upon_creation=False
)

hello_task = BashOperator(
    task_id='test_task',
    bash_command='echo "Hello World from Airflow!"',
    do_xcom_push = True,
    dag=dag,
)

hello_task

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

[atul-astronomer]

set() method is not implemented in task_sdk variable.py

[jason810496]

I can take on this issue and will assign it to myself.

Before I start, I just want to confirm—can I use https://github.com/apache/airflow/pull/46869/files as a reference to port the set method to Variable in TaskSDK?

cc @ashb

---

Update: also reference https://github.com/apache/airflow/pull/44605/files

[amoghrajesh]

@jason810496 yes you should be able to take that one as reference, specifically the Variable.get() method.

[kaxil]

closed by #49005

[jscheffl]

I just did a regression with current main, assuming that #49005 fixes the problem - using EdgeExecutor, LocalExecutor and DAG providers/edge/src/airflow/providers/edge/example_dags/integration_test.py I still get a consistent error:

[2025-04-10, 22:46:22] INFO - DAG bundles loaded: dags-folder, example_dags: source="airflow.dag_processing.bundles.manager.DagBundlesManager"
[2025-04-10, 22:46:22] INFO - Filling up the DagBag from /files/dags/integration_test.py: source="airflow.models.dagbag.DagBag"
[2025-04-10, 22:46:22] ERROR - Task failed with exception: source="task"
RuntimeError: Direct database access via the ORM is not allowed in Airflow 3.0
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 775 in run
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 1038 in _execute_task
File "/opt/airflow/task-sdk/src/airflow/sdk/bases/operator.py", line 409 in wrapper
File "/opt/airflow/task-sdk/src/airflow/sdk/bases/decorator.py", line 251 in execute
File "/opt/airflow/task-sdk/src/airflow/sdk/bases/operator.py", line 409 in wrapper
File "/opt/airflow/providers/standard/src/airflow/providers/standard/operators/python.py", line 212 in execute
File "/opt/airflow/providers/standard/src/airflow/providers/standard/operators/python.py", line 235 in execute_callable
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/callback_runner.py", line 81 in run
File "/files/dags/integration_test.py", line 101 in variable
File "/opt/airflow/airflow-core/src/airflow/utils/session.py", line 101 in wrapper
File "/usr/local/lib/python3.12/contextlib.py", line 137 in __enter__
File "/opt/airflow/airflow-core/src/airflow/utils/session.py", line 41 in create_session
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/supervisor.py", line 215 in __init__
[2025-04-10, 22:46:22] INFO - Task instance is in running state: chan="stdout": source="task"
[2025-04-10, 22:46:22] INFO -  Previous state of the Task instance: queued: chan="stdout": source="task"
[2025-04-10, 22:46:22] INFO - Current task name:variable: chan="stdout": source="task"
[2025-04-10, 22:46:22] INFO - Dag name:integration_test: chan="stdout": source="task"

DAG code snippet:

    @task
    def variable():
        Variable.set("integration_test_key", "value")
        if Variable.get("integration_test_key") != "value":
            raise ValueError("Variable not set as expected.")
        Variable.delete("integration_test_key")