DagRun.find() throws an "Direct database access via the ORM is not allowed in Airflow 3.0" error

[Michael-cd30]

Apache Airflow version

3.0.1

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

No response

What happened?

A call to DagRun.find() throws the following RuntimeError :
Direct database access via the ORM is not allowed in Airflow 3.0

What you think should happen instead?

According to the 3.0.1 documentation, this function should return the set of DAG runs that matches the search criteria.

How to reproduce

Run this DAG

from airflow.decorators import dag, task
from airflow.models import DagRun
from airflow.utils.state import DagRunState
from datetime import datetime

@dag(dag_id = "test_airflow3",
     start_date = datetime(2025, 1, 1),
     schedule = "@hourly",
     tags = ["misc"],
     catchup = False)
def mon_dag():

    @task(task_id = "appel_dagrun_find")
    def ma_premiere_tache():
        print('DagRun.find(dag_id = "test_airflow3")')
        print(DagRun.find(dag_id = "test_airflow3"))
        

    ma_premiere_tache()

mon_dag()

Operating System

PRETTY_NAME="Debian GNU/Linux 12 (bookworm)"

Versions of Apache Airflow Providers

apache-airflow-providers-amazon==9.7.0
apache-airflow-providers-celery==3.10.6
apache-airflow-providers-cncf-kubernetes==10.4.3
apache-airflow-providers-common-compat==1.6.1
apache-airflow-providers-common-io==1.5.4
apache-airflow-providers-common-messaging==1.0.1
apache-airflow-providers-common-sql==1.27.0
apache-airflow-providers-docker==4.3.1
apache-airflow-providers-elasticsearch==6.2.2
apache-airflow-providers-fab==2.0.2
apache-airflow-providers-ftp==3.12.3
apache-airflow-providers-git==0.0.2
apache-airflow-providers-google==15.1.0
apache-airflow-providers-grpc==3.7.3
apache-airflow-providers-hashicorp==4.1.1
apache-airflow-providers-http==5.2.2
apache-airflow-providers-microsoft-azure==12.3.1
apache-airflow-providers-mysql==6.2.2
apache-airflow-providers-odbc==4.9.2
apache-airflow-providers-openlineage==2.2.0
apache-airflow-providers-oracle==4.1.0
apache-airflow-providers-postgres==6.1.3
apache-airflow-providers-redis==4.0.2
apache-airflow-providers-sendgrid==4.0.1
apache-airflow-providers-sftp==5.2.1
apache-airflow-providers-slack==9.0.5
apache-airflow-providers-smtp==2.0.3
apache-airflow-providers-snowflake==6.3.0
apache-airflow-providers-ssh==4.0.1
apache-airflow-providers-standard==1.1.0

Deployment

Docker-Compose

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

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[jroachgolf84]

I was able to reproduce this exception using the logic provided above.

[kevinhongzl]

@jroachgolf84 I'd like to help with this issue :)

[kaxil]

This is not an issue, it is by design that in Airflow 3.0+ you cannot use Airflow metadata DB in your task. You should only use anything in airflow.sdk namespace.

Reference:

• https://airflow.apache.org/docs/apache-airflow/stable/release_notes.html#restricted-metadata-database-access
• https://airflow.apache.org/docs/apache-airflow/stable/release_notes.html#task-execution-api-task-sdk-aip-72
• https://airflow.apache.org/docs/apache-airflow/stable/release_notes.html#new-stable-dag-authoring-interface-airflow-sdk
• https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-72+Task+Execution+Interface+aka+Task+SDK

Use official REST API or Python Client instead.

[kaxil]

Depending on what you want to do, you can also use Task Context:

from airflow.sdk import dag, get_current_context, task
from airflow.utils.state import DagRunState
from datetime import datetime

@dag(dag_id = "test_airflow3",
     start_date = datetime(2025, 1, 1),
     schedule = "@hourly",
     tags = ["misc"],
     catchup = False)
def mon_dag():

    @task(task_id = "appel_dagrun_find")
    def ma_premiere_tache():
        context = get_current_context()
        ti = context["ti"]
        dag_run = context["dag_run"]        

    ma_premiere_tache()

mon_dag()

and then other options:

• ti.get_dr_count
• ti.get_dagrun_state

airflow/task-sdk/src/airflow/sdk/types.py

Lines 100 to 118 in bf0bfe9

	def get_task_states(
	dag_id: str,
	map_index: int | None = None,
	task_ids: list[str] | None = None,
	task_group_id: str | None = None,
	logical_dates: list[AwareDatetime] | None = None,
	run_ids: list[str] | None = None,
	) -> dict[str, Any]: ...
	@staticmethod
	def get_dr_count(
	dag_id: str,
	logical_dates: list[AwareDatetime] | None = None,
	run_ids: list[str] | None = None,
	states: list[str] | None = None,
	) -> int: ...
	@staticmethod
	def get_dagrun_state(dag_id: str, run_id: str) -> str: ...

[Michael-cd30]

The need is to know whether DAGs with specific tags are running or not.

We use this information to synchronize the DAGs.

For example, each "transform" DAG will wait for the "extract_load" DAGs to finish.

Therfore, I don't think the task_context will help.

I think that if this isn't a bug, the documentation should at least be updated.

[kaxil]

The need is to know whether DAGs with specific tags are running or not.

We use this information to synchronize the DAGs.

For example, each "transform" DAG will wait for the "extract_load" DAGs to finish.

Therfore, I don't think the task_context will help.

Yeah, for that use-case Task Context won't help. The option then is using official REST API or Python Client instead.

I think that if this isn't a bug, the documentation should at least be updated.

Yup, it is WIP