session.query is no longer supported in airflow 3.0.2

[krzysztofstepien-tomtom]

Apache Airflow version

3.0.2

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

No response

What happened?

Hi

I am currently migrating my project from airflow 2.10.5 to airflow 3.0.2

Inside i've already have a service which manages dags state i.e

def _get_dag_run_state(context: Any) -> Optional[DagRunState]:
  dag_run_id = context["dag_run"].id
  event_body = Params.deserialize(context["params"], 0).event_body
  product = f"{event_body.product_line}_{event_body.product_version}"

  dag_run_state = None
  session = Session()
  try:
    state_record: DagRunMetadata = session.query(DagRunMetadata).filter_by(dag_run_id=dag_run_id, product=product).first()
    dag_run_state = state_record.state if state_record else DagRunState.UNKNOWN
  except Exception as ex:
    logging.error(f"Failed to get DagRun state for `{product}`.\n Error: {ex}")
  finally:
    session.close()
  return dag_run_state

However after trying to launch code in the latest airflow i am getting the following problem:
RuntimeError: Direct database access via the ORM is not allowed in Airflow 3.0

Is there any workaround in the latest airflow to manage dags state?

What you think should happen instead?

No response

How to reproduce

Any usage of session.query

Operating System

macOS Sequoia 15.5

Versions of Apache Airflow Providers

apache-airflow-providers-fab==2.2.1
apache-airflow-providers-cncf-kubernetes==8.4.0
apache-airflow-providers-databricks==7.4.0
apache-airflow-providers-microsoft-azure==12.4.1
apache-airflow-providers-standard==1.3.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[bot]]

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.

[potiuk]

Correct. Direct database access possibility has been strongly discouraged from the beginning of Airflow 2, it has been mentioned as not part of Public Interface and removed in Airflow 3 . You should use task-sdk API or REST API to interact with Airflow data model

[potiuk]

Converting it to a discussion if more is needed.

[krzysztofstepien-tomtom]

Thank you for the reply!
What do mean by task-sdk? Could you share some examples? Do you mean using f.e Variable?