We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Improve flow run output
**Describe the bug**
The current output of `flow run` is not compatible with what was specified in the original ticket #942.

**Version**
* SQL CLI: [0.0.7]

**To Reproduce**
```
$ flow run example_templating --env dev --project-dir=final

Created dagrun <DagRun example_templating @ 2022-10-19T13:25:14.278881+00:00: 
manual__2022-10-19T13:25:14.278881+00:00, state:running, queued_at: None. externally triggered: False> <DagRun 
example_templating @ 2022-10-19T13:25:14.278881+00:00: manual__2022-10-19T13:25:14.278881+00:00, state:running, 
queued_at: None. externally triggered: False>
*****************************************************
Running task filtered_orders
filtered_orders ran successfully!
*****************************************************
*****************************************************
Running task joint_orders_customers
joint_orders_customers ran successfully!
*****************************************************
Dagrun example_templating final state: success

```

**Expected behaviour**
Print something like:
```
Running the workflow example_imdb…

Processing popular_movies.sql to output_db.popular_movies SUCCESS 
Processing top_directors.sql to output_db.top_directors SUCCESS 
Completed successfully running the pipeline example_imdb
Total time:        3.218s
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `sql-cli/sql_cli/run_dag.py`
Content:
```
1 from __future__ import annotations
2 
3 import logging
4 import sys
5 import warnings
6 from datetime import datetime
7 from typing import Any
8 
9 from airflow.models.connection import Connection
10 from airflow.models.dag import DAG
11 from airflow.models.dagrun import DagRun
12 from airflow.models.taskinstance import TaskInstance
13 from airflow.secrets.local_filesystem import LocalFilesystemBackend
14 from airflow.utils import timezone
15 from airflow.utils.session import NEW_SESSION, provide_session
16 from airflow.utils.state import DagRunState, State
17 from airflow.utils.types import DagRunType
18 from rich import print as pprint
19 from sqlalchemy.orm.session import Session
20 
21 from astro.sql.operators.cleanup import AstroCleanupException
22 
23 log = logging.getLogger(__name__)
24 log.setLevel(logging.INFO)
25 
26 
27 class AstroFilesystemBackend(LocalFilesystemBackend):
28     def __init__(
29         self,
30         connections: dict[str, Connection] | None = None,
31         variables_file_path: str | None = None,
32         connections_file_path: str | None = None,
33     ):
34         self._local_conns: dict[str, Connection] = connections or {}
35         super().__init__(variables_file_path=variables_file_path, connections_file_path=connections_file_path)
36 
37     @property
38     def _local_connections(self) -> dict[str, Connection]:
39         conns = self._local_conns
40         conns.update(super()._local_connections)
41         return conns
42 
43 
44 @provide_session
45 def run_dag(
46     dag: DAG,
47     execution_date: datetime | None = None,
48     run_conf: dict[str, Any] | None = None,
49     conn_file_path: str | None = None,
50     variable_file_path: str | None = None,
51     connections: dict[str, Connection] | None = None,
52     verbose: bool = False,
53     session: Session = NEW_SESSION,
54 ) -> None:
55     """
56     Execute one single DagRun for a given DAG and execution date.
57 
58     :param dag: The Airflow DAG we will run
59     :param execution_date: execution date for the DAG run
60     :param run_conf: configuration to pass to newly created dagrun
61     :param conn_file_path: file path to a connection file in either yaml or json
62     :param variable_file_path: file path to a variable file in either yaml or json
63     :param session: database connection (optional)
64     """
65 
66     execution_date = execution_date or timezone.utcnow()
67     dag.log.debug("Clearing existing task instances for execution date %s", execution_date)
68     dag.clear(
69         start_date=execution_date,
70         end_date=execution_date,
71         dag_run_state=False,  # type: ignore
72         session=session,
73     )
74     dag.log.debug("Getting dagrun for dag %s", dag.dag_id)
75     dr: DagRun = _get_or_create_dagrun(
76         dag=dag,
77         start_date=execution_date,
78         execution_date=execution_date,
79         run_id=DagRun.generate_run_id(DagRunType.MANUAL, execution_date),
80         session=session,
81         conf=run_conf,
82     )
83 
84     if conn_file_path or variable_file_path or connections:
85         # To get around the fact that we reload certain modules, we need to import here
86         from airflow.configuration import secrets_backend_list
87 
88         local_secrets = AstroFilesystemBackend(
89             variables_file_path=variable_file_path,
90             connections_file_path=conn_file_path,
91             connections=connections,
92         )
93         secrets_backend_list.insert(0, local_secrets)
94 
95     tasks = dag.task_dict
96     dag.log.debug("starting dagrun")
97     # Instead of starting a scheduler, we run the minimal loop possible to check
98     # for task readiness and dependency management. This is notably faster
99     # than creating a BackfillJob and allows us to surface logs to the user
100     while dr.state == State.RUNNING:
101         schedulable_tis, _ = dr.update_state(session=session)
102         for ti in schedulable_tis:
103             if verbose:
104                 add_logger_if_needed(dag, ti)
105             ti.task = tasks[ti.task_id]
106             _run_task(ti, session=session)
107     pprint(f"Dagrun {dr.dag_id} final state: {dr.state}")
108     if conn_file_path or variable_file_path or connections:
109         # Remove the local variables we have added to the secrets_backend_list
110         secrets_backend_list.pop(0)  # noqa
111 
112 
113 def add_logger_if_needed(dag: DAG, ti: TaskInstance) -> None:
114     """
115     Add a formatted logger to the taskinstance so all logs are surfaced to the command line instead
116     of into a task file. Since this is a local test run, it is much better for the user to see logs
117     in the command line, rather than needing to search for a log file.
118     :param ti: The taskinstance that will receive a logger
119 
120     """
121     logging_format = logging.Formatter("[%(asctime)s] {%(filename)s:%(lineno)d} %(levelname)s - %(message)s")
122     handler = logging.StreamHandler(sys.stdout)
123     handler.level = logging.DEBUG
124     handler.setFormatter(logging_format)
125     # only add log handler once
126     if not any(isinstance(h, logging.StreamHandler) for h in ti.log.handlers):
127         dag.log.debug("Adding Streamhandler to taskinstance %s", ti.task_id)
128         ti.log.addHandler(handler)
129 
130 
131 def _run_task(ti: TaskInstance, session: Session) -> None:
132     """
133     Run a single task instance, and push result to Xcom for downstream tasks. Bypasses a lot of
134     extra steps used in `task.run` to keep our local running as fast as possible
135     This function is only meant for the `dag.test` function as a helper function.
136 
137     :param ti: TaskInstance to run
138     """
139     pprint("[bold green]*****************************************************[/bold green]")
140     if hasattr(ti, "map_index") and ti.map_index > 0:
141         pprint("Running task %s index %d", ti.task_id, ti.map_index)
142     else:
143         pprint(f"Running task [bold red]{ti.task_id}[/bold red]")
144     try:
145         warnings.filterwarnings(action="ignore")
146         ti._run_raw_task(session=session)  # skipcq: PYL-W0212
147         session.flush()
148         session.commit()
149         pprint(f"[bold red]{ti.task_id}[/bold red] ran successfully!")
150     except AstroCleanupException:
151         pprint("aql.cleanup async, continuing")
152     pprint("[bold green]*****************************************************[/bold green]")
153 
154 
155 def _get_or_create_dagrun(
156     dag: DAG,
157     conf: dict[Any, Any] | None,
158     start_date: datetime,
159     execution_date: datetime,
160     run_id: str,
161     session: Session,
162 ) -> DagRun:
163     """
164     Create a DAGRun, but only after clearing the previous instance of said dagrun to prevent collisions.
165     This function is only meant for the `dag.test` function as a helper function.
166     :param dag: Dag to be used to find dagrun
167     :param conf: configuration to pass to newly created dagrun
168     :param start_date: start date of new dagrun, defaults to execution_date
169     :param execution_date: execution_date for finding the dagrun
170     :param run_id: run_id to pass to new dagrun
171     :param session: sqlalchemy session
172     :return: the Dagrun object needed to run tasks.
173     """
174     log.info("dagrun id: %s", dag.dag_id)
175     dr: DagRun = (
176         session.query(DagRun)
177         .filter(DagRun.dag_id == dag.dag_id, DagRun.execution_date == execution_date)
178         .first()
179     )
180     if dr:
181         session.delete(dr)
182         session.commit()
183     dr = dag.create_dagrun(
184         state=DagRunState.RUNNING,
185         execution_date=execution_date,
186         run_id=run_id,
187         start_date=start_date or execution_date,
188         session=session,
189         conf=conf,  # type: ignore
190     )
191     pprint(f"Created dagrun [bold blue]{str(dr)}[/bold blue]", str(dr))
192     return dr
193
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/sql-cli/sql_cli/run_dag.py b/sql-cli/sql_cli/run_dag.py
--- a/sql-cli/sql_cli/run_dag.py
+++ b/sql-cli/sql_cli/run_dag.py
@@ -104,7 +104,9 @@
                 add_logger_if_needed(dag, ti)
             ti.task = tasks[ti.task_id]
             _run_task(ti, session=session)
-    pprint(f"Dagrun {dr.dag_id} final state: {dr.state}")
+    pprint(f"Completed running the workflow {dr.dag_id}: [bold yellow][{dr.state.upper()}][/bold yellow]")
+    elapsed_seconds = (dr.end_date - dr.start_date).microseconds / 10**6
+    pprint(f"Total elapsed time: [bold blue]{elapsed_seconds:.2}s[/bold blue]")
     if conn_file_path or variable_file_path or connections:
         # Remove the local variables we have added to the secrets_backend_list
         secrets_backend_list.pop(0)  # noqa
@@ -136,20 +138,19 @@
 
     :param ti: TaskInstance to run
     """
-    pprint("[bold green]*****************************************************[/bold green]")
-    if hasattr(ti, "map_index") and ti.map_index > 0:
-        pprint("Running task %s index %d", ti.task_id, ti.map_index)
+    if ti.map_index >= 0:
+        pprint(f"Processing [bold yellow]{ti.task_id}[/bold yellow][{ti.map_index}]...", end=" ")
     else:
-        pprint(f"Running task [bold red]{ti.task_id}[/bold red]")
+        pprint(f"Processing [bold yellow]{ti.task_id}[/bold yellow]...", end=" ")
+
     try:
         warnings.filterwarnings(action="ignore")
         ti._run_raw_task(session=session)  # skipcq: PYL-W0212
         session.flush()
         session.commit()
-        pprint(f"[bold red]{ti.task_id}[/bold red] ran successfully!")
+        pprint("[bold green]SUCCESS[/bold green]")
     except AstroCleanupException:
         pprint("aql.cleanup async, continuing")
-    pprint("[bold green]*****************************************************[/bold green]")
 
 
 def _get_or_create_dagrun(
@@ -188,5 +189,5 @@
         session=session,
         conf=conf,  # type: ignore
     )
-    pprint(f"Created dagrun [bold blue]{str(dr)}[/bold blue]", str(dr))
+    pprint(f"\nRunning the workflow [bold blue]{dag.dag_id}[/bold blue]\n")
     return dr

{"golden_diff": "diff --git a/sql-cli/sql_cli/run_dag.py b/sql-cli/sql_cli/run_dag.py\n--- a/sql-cli/sql_cli/run_dag.py\n+++ b/sql-cli/sql_cli/run_dag.py\n@@ -104,7 +104,9 @@\n                 add_logger_if_needed(dag, ti)\n             ti.task = tasks[ti.task_id]\n             _run_task(ti, session=session)\n-    pprint(f\"Dagrun {dr.dag_id} final state: {dr.state}\")\n+    pprint(f\"Completed running the workflow {dr.dag_id}: [bold yellow][{dr.state.upper()}][/bold yellow]\")\n+    elapsed_seconds = (dr.end_date - dr.start_date).microseconds / 10**6\n+    pprint(f\"Total elapsed time: [bold blue]{elapsed_seconds:.2}s[/bold blue]\")\n     if conn_file_path or variable_file_path or connections:\n         # Remove the local variables we have added to the secrets_backend_list\n         secrets_backend_list.pop(0)  # noqa\n@@ -136,20 +138,19 @@\n \n     :param ti: TaskInstance to run\n     \"\"\"\n-    pprint(\"[bold green]*****************************************************[/bold green]\")\n-    if hasattr(ti, \"map_index\") and ti.map_index > 0:\n-        pprint(\"Running task %s index %d\", ti.task_id, ti.map_index)\n+    if ti.map_index >= 0:\n+        pprint(f\"Processing [bold yellow]{ti.task_id}[/bold yellow][{ti.map_index}]...\", end=\" \")\n     else:\n-        pprint(f\"Running task [bold red]{ti.task_id}[/bold red]\")\n+        pprint(f\"Processing [bold yellow]{ti.task_id}[/bold yellow]...\", end=\" \")\n+\n     try:\n         warnings.filterwarnings(action=\"ignore\")\n         ti._run_raw_task(session=session)  # skipcq: PYL-W0212\n         session.flush()\n         session.commit()\n-        pprint(f\"[bold red]{ti.task_id}[/bold red] ran successfully!\")\n+        pprint(\"[bold green]SUCCESS[/bold green]\")\n     except AstroCleanupException:\n         pprint(\"aql.cleanup async, continuing\")\n-    pprint(\"[bold green]*****************************************************[/bold green]\")\n \n \n def _get_or_create_dagrun(\n@@ -188,5 +189,5 @@\n         session=session,\n         conf=conf,  # type: ignore\n     )\n-    pprint(f\"Created dagrun [bold blue]{str(dr)}[/bold blue]\", str(dr))\n+    pprint(f\"\\nRunning the workflow [bold blue]{dag.dag_id}[/bold blue]\\n\")\n     return dr\n", "issue": "Improve flow run output\n**Describe the bug**\r\nThe current output of `flow run` is not compatible with what was specified in the original ticket #942.\r\n\r\n**Version**\r\n* SQL CLI: [0.0.7]\r\n\r\n**To Reproduce**\r\n```\r\n$ flow run example_templating --env dev --project-dir=final\r\n\r\nCreated dagrun <DagRun example_templating @ 2022-10-19T13:25:14.278881+00:00: \r\nmanual__2022-10-19T13:25:14.278881+00:00, state:running, queued_at: None. externally triggered: False> <DagRun \r\nexample_templating @ 2022-10-19T13:25:14.278881+00:00: manual__2022-10-19T13:25:14.278881+00:00, state:running, \r\nqueued_at: None. externally triggered: False>\r\n*****************************************************\r\nRunning task filtered_orders\r\nfiltered_orders ran successfully!\r\n*****************************************************\r\n*****************************************************\r\nRunning task joint_orders_customers\r\njoint_orders_customers ran successfully!\r\n*****************************************************\r\nDagrun example_templating final state: success\r\n\r\n```\r\n\r\n**Expected behaviour**\r\nPrint something like:\r\n```\r\nRunning the workflow example_imdb\u2026\r\n\r\nProcessing popular_movies.sql to output_db.popular_movies SUCCESS \r\nProcessing top_directors.sql to output_db.top_directors SUCCESS \r\nCompleted successfully running the pipeline example_imdb\r\nTotal time:        3.218s\r\n```\n", "before_files": [{"content": "from __future__ import annotations\n\nimport logging\nimport sys\nimport warnings\nfrom datetime import datetime\nfrom typing import Any\n\nfrom airflow.models.connection import Connection\nfrom airflow.models.dag import DAG\nfrom airflow.models.dagrun import DagRun\nfrom airflow.models.taskinstance import TaskInstance\nfrom airflow.secrets.local_filesystem import LocalFilesystemBackend\nfrom airflow.utils import timezone\nfrom airflow.utils.session import NEW_SESSION, provide_session\nfrom airflow.utils.state import DagRunState, State\nfrom airflow.utils.types import DagRunType\nfrom rich import print as pprint\nfrom sqlalchemy.orm.session import Session\n\nfrom astro.sql.operators.cleanup import AstroCleanupException\n\nlog = logging.getLogger(__name__)\nlog.setLevel(logging.INFO)\n\n\nclass AstroFilesystemBackend(LocalFilesystemBackend):\n    def __init__(\n        self,\n        connections: dict[str, Connection] | None = None,\n        variables_file_path: str | None = None,\n        connections_file_path: str | None = None,\n    ):\n        self._local_conns: dict[str, Connection] = connections or {}\n        super().__init__(variables_file_path=variables_file_path, connections_file_path=connections_file_path)\n\n    @property\n    def _local_connections(self) -> dict[str, Connection]:\n        conns = self._local_conns\n        conns.update(super()._local_connections)\n        return conns\n\n\n@provide_session\ndef run_dag(\n    dag: DAG,\n    execution_date: datetime | None = None,\n    run_conf: dict[str, Any] | None = None,\n    conn_file_path: str | None = None,\n    variable_file_path: str | None = None,\n    connections: dict[str, Connection] | None = None,\n    verbose: bool = False,\n    session: Session = NEW_SESSION,\n) -> None:\n    \"\"\"\n    Execute one single DagRun for a given DAG and execution date.\n\n    :param dag: The Airflow DAG we will run\n    :param execution_date: execution date for the DAG run\n    :param run_conf: configuration to pass to newly created dagrun\n    :param conn_file_path: file path to a connection file in either yaml or json\n    :param variable_file_path: file path to a variable file in either yaml or json\n    :param session: database connection (optional)\n    \"\"\"\n\n    execution_date = execution_date or timezone.utcnow()\n    dag.log.debug(\"Clearing existing task instances for execution date %s\", execution_date)\n    dag.clear(\n        start_date=execution_date,\n        end_date=execution_date,\n        dag_run_state=False,  # type: ignore\n        session=session,\n    )\n    dag.log.debug(\"Getting dagrun for dag %s\", dag.dag_id)\n    dr: DagRun = _get_or_create_dagrun(\n        dag=dag,\n        start_date=execution_date,\n        execution_date=execution_date,\n        run_id=DagRun.generate_run_id(DagRunType.MANUAL, execution_date),\n        session=session,\n        conf=run_conf,\n    )\n\n    if conn_file_path or variable_file_path or connections:\n        # To get around the fact that we reload certain modules, we need to import here\n        from airflow.configuration import secrets_backend_list\n\n        local_secrets = AstroFilesystemBackend(\n            variables_file_path=variable_file_path,\n            connections_file_path=conn_file_path,\n            connections=connections,\n        )\n        secrets_backend_list.insert(0, local_secrets)\n\n    tasks = dag.task_dict\n    dag.log.debug(\"starting dagrun\")\n    # Instead of starting a scheduler, we run the minimal loop possible to check\n    # for task readiness and dependency management. This is notably faster\n    # than creating a BackfillJob and allows us to surface logs to the user\n    while dr.state == State.RUNNING:\n        schedulable_tis, _ = dr.update_state(session=session)\n        for ti in schedulable_tis:\n            if verbose:\n                add_logger_if_needed(dag, ti)\n            ti.task = tasks[ti.task_id]\n            _run_task(ti, session=session)\n    pprint(f\"Dagrun {dr.dag_id} final state: {dr.state}\")\n    if conn_file_path or variable_file_path or connections:\n        # Remove the local variables we have added to the secrets_backend_list\n        secrets_backend_list.pop(0)  # noqa\n\n\ndef add_logger_if_needed(dag: DAG, ti: TaskInstance) -> None:\n    \"\"\"\n    Add a formatted logger to the taskinstance so all logs are surfaced to the command line instead\n    of into a task file. Since this is a local test run, it is much better for the user to see logs\n    in the command line, rather than needing to search for a log file.\n    :param ti: The taskinstance that will receive a logger\n\n    \"\"\"\n    logging_format = logging.Formatter(\"[%(asctime)s] {%(filename)s:%(lineno)d} %(levelname)s - %(message)s\")\n    handler = logging.StreamHandler(sys.stdout)\n    handler.level = logging.DEBUG\n    handler.setFormatter(logging_format)\n    # only add log handler once\n    if not any(isinstance(h, logging.StreamHandler) for h in ti.log.handlers):\n        dag.log.debug(\"Adding Streamhandler to taskinstance %s\", ti.task_id)\n        ti.log.addHandler(handler)\n\n\ndef _run_task(ti: TaskInstance, session: Session) -> None:\n    \"\"\"\n    Run a single task instance, and push result to Xcom for downstream tasks. Bypasses a lot of\n    extra steps used in `task.run` to keep our local running as fast as possible\n    This function is only meant for the `dag.test` function as a helper function.\n\n    :param ti: TaskInstance to run\n    \"\"\"\n    pprint(\"[bold green]*****************************************************[/bold green]\")\n    if hasattr(ti, \"map_index\") and ti.map_index > 0:\n        pprint(\"Running task %s index %d\", ti.task_id, ti.map_index)\n    else:\n        pprint(f\"Running task [bold red]{ti.task_id}[/bold red]\")\n    try:\n        warnings.filterwarnings(action=\"ignore\")\n        ti._run_raw_task(session=session)  # skipcq: PYL-W0212\n        session.flush()\n        session.commit()\n        pprint(f\"[bold red]{ti.task_id}[/bold red] ran successfully!\")\n    except AstroCleanupException:\n        pprint(\"aql.cleanup async, continuing\")\n    pprint(\"[bold green]*****************************************************[/bold green]\")\n\n\ndef _get_or_create_dagrun(\n    dag: DAG,\n    conf: dict[Any, Any] | None,\n    start_date: datetime,\n    execution_date: datetime,\n    run_id: str,\n    session: Session,\n) -> DagRun:\n    \"\"\"\n    Create a DAGRun, but only after clearing the previous instance of said dagrun to prevent collisions.\n    This function is only meant for the `dag.test` function as a helper function.\n    :param dag: Dag to be used to find dagrun\n    :param conf: configuration to pass to newly created dagrun\n    :param start_date: start date of new dagrun, defaults to execution_date\n    :param execution_date: execution_date for finding the dagrun\n    :param run_id: run_id to pass to new dagrun\n    :param session: sqlalchemy session\n    :return: the Dagrun object needed to run tasks.\n    \"\"\"\n    log.info(\"dagrun id: %s\", dag.dag_id)\n    dr: DagRun = (\n        session.query(DagRun)\n        .filter(DagRun.dag_id == dag.dag_id, DagRun.execution_date == execution_date)\n        .first()\n    )\n    if dr:\n        session.delete(dr)\n        session.commit()\n    dr = dag.create_dagrun(\n        state=DagRunState.RUNNING,\n        execution_date=execution_date,\n        run_id=run_id,\n        start_date=start_date or execution_date,\n        session=session,\n        conf=conf,  # type: ignore\n    )\n    pprint(f\"Created dagrun [bold blue]{str(dr)}[/bold blue]\", str(dr))\n    return dr\n", "path": "sql-cli/sql_cli/run_dag.py"}], "after_files": [{"content": "from __future__ import annotations\n\nimport logging\nimport sys\nimport warnings\nfrom datetime import datetime\nfrom typing import Any\n\nfrom airflow.models.connection import Connection\nfrom airflow.models.dag import DAG\nfrom airflow.models.dagrun import DagRun\nfrom airflow.models.taskinstance import TaskInstance\nfrom airflow.secrets.local_filesystem import LocalFilesystemBackend\nfrom airflow.utils import timezone\nfrom airflow.utils.session import NEW_SESSION, provide_session\nfrom airflow.utils.state import DagRunState, State\nfrom airflow.utils.types import DagRunType\nfrom rich import print as pprint\nfrom sqlalchemy.orm.session import Session\n\nfrom astro.sql.operators.cleanup import AstroCleanupException\n\nlog = logging.getLogger(__name__)\nlog.setLevel(logging.INFO)\n\n\nclass AstroFilesystemBackend(LocalFilesystemBackend):\n    def __init__(\n        self,\n        connections: dict[str, Connection] | None = None,\n        variables_file_path: str | None = None,\n        connections_file_path: str | None = None,\n    ):\n        self._local_conns: dict[str, Connection] = connections or {}\n        super().__init__(variables_file_path=variables_file_path, connections_file_path=connections_file_path)\n\n    @property\n    def _local_connections(self) -> dict[str, Connection]:\n        conns = self._local_conns\n        conns.update(super()._local_connections)\n        return conns\n\n\n@provide_session\ndef run_dag(\n    dag: DAG,\n    execution_date: datetime | None = None,\n    run_conf: dict[str, Any] | None = None,\n    conn_file_path: str | None = None,\n    variable_file_path: str | None = None,\n    connections: dict[str, Connection] | None = None,\n    verbose: bool = False,\n    session: Session = NEW_SESSION,\n) -> None:\n    \"\"\"\n    Execute one single DagRun for a given DAG and execution date.\n\n    :param dag: The Airflow DAG we will run\n    :param execution_date: execution date for the DAG run\n    :param run_conf: configuration to pass to newly created dagrun\n    :param conn_file_path: file path to a connection file in either yaml or json\n    :param variable_file_path: file path to a variable file in either yaml or json\n    :param session: database connection (optional)\n    \"\"\"\n\n    execution_date = execution_date or timezone.utcnow()\n    dag.log.debug(\"Clearing existing task instances for execution date %s\", execution_date)\n    dag.clear(\n        start_date=execution_date,\n        end_date=execution_date,\n        dag_run_state=False,  # type: ignore\n        session=session,\n    )\n    dag.log.debug(\"Getting dagrun for dag %s\", dag.dag_id)\n    dr: DagRun = _get_or_create_dagrun(\n        dag=dag,\n        start_date=execution_date,\n        execution_date=execution_date,\n        run_id=DagRun.generate_run_id(DagRunType.MANUAL, execution_date),\n        session=session,\n        conf=run_conf,\n    )\n\n    if conn_file_path or variable_file_path or connections:\n        # To get around the fact that we reload certain modules, we need to import here\n        from airflow.configuration import secrets_backend_list\n\n        local_secrets = AstroFilesystemBackend(\n            variables_file_path=variable_file_path,\n            connections_file_path=conn_file_path,\n            connections=connections,\n        )\n        secrets_backend_list.insert(0, local_secrets)\n\n    tasks = dag.task_dict\n    dag.log.debug(\"starting dagrun\")\n    # Instead of starting a scheduler, we run the minimal loop possible to check\n    # for task readiness and dependency management. This is notably faster\n    # than creating a BackfillJob and allows us to surface logs to the user\n    while dr.state == State.RUNNING:\n        schedulable_tis, _ = dr.update_state(session=session)\n        for ti in schedulable_tis:\n            if verbose:\n                add_logger_if_needed(dag, ti)\n            ti.task = tasks[ti.task_id]\n            _run_task(ti, session=session)\n    pprint(f\"Completed running the workflow {dr.dag_id}: [bold yellow][{dr.state.upper()}][/bold yellow]\")\n    elapsed_seconds = (dr.end_date - dr.start_date).microseconds / 10**6\n    pprint(f\"Total elapsed time: [bold blue]{elapsed_seconds:.2}s[/bold blue]\")\n    if conn_file_path or variable_file_path or connections:\n        # Remove the local variables we have added to the secrets_backend_list\n        secrets_backend_list.pop(0)  # noqa\n\n\ndef add_logger_if_needed(dag: DAG, ti: TaskInstance) -> None:\n    \"\"\"\n    Add a formatted logger to the taskinstance so all logs are surfaced to the command line instead\n    of into a task file. Since this is a local test run, it is much better for the user to see logs\n    in the command line, rather than needing to search for a log file.\n    :param ti: The taskinstance that will receive a logger\n\n    \"\"\"\n    logging_format = logging.Formatter(\"[%(asctime)s] {%(filename)s:%(lineno)d} %(levelname)s - %(message)s\")\n    handler = logging.StreamHandler(sys.stdout)\n    handler.level = logging.DEBUG\n    handler.setFormatter(logging_format)\n    # only add log handler once\n    if not any(isinstance(h, logging.StreamHandler) for h in ti.log.handlers):\n        dag.log.debug(\"Adding Streamhandler to taskinstance %s\", ti.task_id)\n        ti.log.addHandler(handler)\n\n\ndef _run_task(ti: TaskInstance, session: Session) -> None:\n    \"\"\"\n    Run a single task instance, and push result to Xcom for downstream tasks. Bypasses a lot of\n    extra steps used in `task.run` to keep our local running as fast as possible\n    This function is only meant for the `dag.test` function as a helper function.\n\n    :param ti: TaskInstance to run\n    \"\"\"\n    if ti.map_index >= 0:\n        pprint(f\"Processing [bold yellow]{ti.task_id}[/bold yellow][{ti.map_index}]...\", end=\" \")\n    else:\n        pprint(f\"Processing [bold yellow]{ti.task_id}[/bold yellow]...\", end=\" \")\n\n    try:\n        warnings.filterwarnings(action=\"ignore\")\n        ti._run_raw_task(session=session)  # skipcq: PYL-W0212\n        session.flush()\n        session.commit()\n        pprint(\"[bold green]SUCCESS[/bold green]\")\n    except AstroCleanupException:\n        pprint(\"aql.cleanup async, continuing\")\n\n\ndef _get_or_create_dagrun(\n    dag: DAG,\n    conf: dict[Any, Any] | None,\n    start_date: datetime,\n    execution_date: datetime,\n    run_id: str,\n    session: Session,\n) -> DagRun:\n    \"\"\"\n    Create a DAGRun, but only after clearing the previous instance of said dagrun to prevent collisions.\n    This function is only meant for the `dag.test` function as a helper function.\n    :param dag: Dag to be used to find dagrun\n    :param conf: configuration to pass to newly created dagrun\n    :param start_date: start date of new dagrun, defaults to execution_date\n    :param execution_date: execution_date for finding the dagrun\n    :param run_id: run_id to pass to new dagrun\n    :param session: sqlalchemy session\n    :return: the Dagrun object needed to run tasks.\n    \"\"\"\n    log.info(\"dagrun id: %s\", dag.dag_id)\n    dr: DagRun = (\n        session.query(DagRun)\n        .filter(DagRun.dag_id == dag.dag_id, DagRun.execution_date == execution_date)\n        .first()\n    )\n    if dr:\n        session.delete(dr)\n        session.commit()\n    dr = dag.create_dagrun(\n        state=DagRunState.RUNNING,\n        execution_date=execution_date,\n        run_id=run_id,\n        start_date=start_date or execution_date,\n        session=session,\n        conf=conf,  # type: ignore\n    )\n    pprint(f\"\\nRunning the workflow [bold blue]{dag.dag_id}[/bold blue]\\n\")\n    return dr\n", "path": "sql-cli/sql_cli/run_dag.py"}]}