code: def dag_state(args, session: Session = NEW_SESSION) -> None:
      """
      Return the state (and conf if exists) of a DagRun at the command line.
  
      >>> airflow dags state tutorial 2015-01-01T00:00:00.000000
      running
      >>> airflow dags state a_dag_with_conf_passed 2015-01-01T00:00:00.000000
      failed, {"name": "bob", "age": "42"}
      """
      dag = DagModel.get_dagmodel(args.dag_id, session=session)
  
      if not dag:
          raise SystemExit(
              f"DAG: {args.dag_id} does not exist in 'dag' table"
          )
      dr = session.scalar(
          select(DagRun).filter_by(
              dag_id=args.dag_id, execution_date=args.execution_date
          )
      )
      out = dr.state if dr else None
      conf_out = ""
      if out and dr.conf:
          conf_out = ", " + json.dumps(dr.conf)
      print(str(out) + conf_out)
  
imports: from __future__ import annotations
  import ast
  import errno
  import json
  import logging
  import operator
  import signal
  import subprocess
  import sys
  import warnings
  from typing import TYPE_CHECKING
  from sqlalchemy import delete, select
  from airflow import settings
  from airflow.api.client import get_current_api_client
  from airflow.api_connexion.schemas.dag_schema import dag_schema
  from airflow.cli.simple_table import AirflowConsole
  from airflow.configuration import conf
  from airflow.exceptions import AirflowException, RemovedInAirflow3Warning
  from airflow.jobs.job import Job
  from airflow.models import DagBag, DagModel, DagRun, TaskInstance
  from airflow.models.dag import DAG
  from airflow.models.serialized_dag import SerializedDagModel
  from airflow.utils import cli, timezone
  from airflow.utils.cli import get_dag, get_dags, process_subdir, sigint_handler, suppress_logs_and_warning
  from airflow.utils.dot_renderer import render_dag, render_dag_dependencies
  from airflow.utils.providers_configuration_loader import providers_configuration_loaded
  from airflow.utils.session import NEW_SESSION, create_session, provide_session
  from airflow.utils.state import DagRunState
label_desc: Use image node_id 13 for calling a global function with example usage: dag_state(args, session) without return types