Is it possible to use a Connection stored in the metadata DB from a python shell

[brki]

Sometimes when I want to investigate something, it's easiest to it in an interactive shell.

In Airflow 2, I could fire up a python shell and run something like:

from airflow.providers.mysql.hooks.mysql import MySqlHook
hook = MySqlHook(mysql_conn_id="my_conn_id")
engine = hook.get_sqlalchemy_engine()
...

But with Airflow 3, I get: airflow.exceptions.AirflowNotFoundException: The conn_id my_conn_id isn't defined

my_conn_id is a Connection defined and stored in my Airflow metadata DB.

Is there a way to make this work with Airflow 3?

[rmaziere]

Hi,
There are a lot of changes with v3.
You have to use the from airflow.providers.common.sql.operators.sql.SQLExecuteQueryOperator insteat of airflow.providers.mysql.
https://airflow.apache.org/docs/apache-airflow-providers-common-sql/stable/index.html

[potiuk]

Generally - in Airflow 3 - see https://airflow.apache.org/docs/apache-airflow/stable/release_notes.html#airflow-3-0-0-2025-04-22 - you are not allowed to use direct DB connection from worker. Your worker is not supposed to connect to the database at all - it uses execution API to retrieve the connection, This is part of the new architecture of Airflow.

So it really depends where and in what context you start the shell. If you start it on scheduler, I think it might work, but only via Connection public API in scheduler, not interactive session as you do.

The interactive session behaviour you described was never a Public interface of Airflow.

You have not described how you start your interactive shell and what stacktrace you get, but generally speaking I think we've never thought about using it this way. If you describe it in detail in a feature request (i.e. allowing to do stuff interactively - with exact proposal on how you would like to intereact with airflow) - I guess we might consider it and turn it into "actual public API" of airflow and we could make it work in specific circumstances. But it should start with describing a general feature for it, what you would like to do, typical operations you think should work etc.

Also this might still change in Airflow 3.2 -> we are workign on completing task isolation and when it is done, you will have limited access to Airflow packages on worker and no task-sdk on scheduler, so behaviour of it might still change.

One of the things with Airflow 2 and the way you use it that people (like you) used some bits of pieces of Airlfow in their own ways and while it worked "accidentally" - it was never part of the public interface of Airflow. This changed in Airflow 3 where a lot of architectural changes made some things to stop working - and the right way of fixing it is to listen to our users, see how they want to use airflow and implement something they can use in a form of maintainable, publc interface.

So if you would like to describe a more complete use case of "interactive use fo some of the airflow components" or similar - that is a good start to have it implemented.

[brki]

Thanks @potiuk .

The DAGs do run on the workers, and they do deal with open connections to DBs. I hear what you are saying though, and think it's a good idea to have a feature request for an interactive shell.

For info, here is some workaround code that I'm currently using. I then call cli_engine("my_conn_id") to get a working sqlalchemy engine for that connection:

from logging import getLogger

from airflow.models.connection import Connection as AirflowConnection
from airflow.providers.common.sql.hooks.sql import DbApiHook
from airflow.providers.mysql.hooks.mysql import MySqlHook
from airflow.settings import Session
from sqlalchemy.engine import Connection, Engine

log = getLogger(__name__)


def cli_get_connection(conn_id: str) -> Connection:
    """
    In Airflow 3, when using a quick one-off script or interactive terminal,
    it does not load user-defined connections from the DB if you try to do
    something like:
        hook = MySqlHook(mysql_conn_id='my_conn_id')
        # it will raise an error that the connection 'my_conn_id' is not found
    This function works around this limitation by manually querying the connection
    from the Airflow metadata DB.
    """
    with Session() as session:
        conn = (
            session
            .query(AirflowConnection)
            .filter(AirflowConnection.conn_id == conn_id)
            .first()
        )
    if not conn:
        log.warning(f"Connection with id '{conn_id}' not found in Airflow metadata DB")
        # Fall back to try getting one with Connection.get(), which will find
        # a connection if it defined in an environment variable.
        conn = AirflowConnection.get(conn_id)
    return conn


def cli_hook(conn_id: str, hook_class=MySqlHook) -> DbApiHook:
    conn = cli_get_connection(conn_id)
    return hook_class(connection=conn)


def cli_engine(conn_id: str, hook_class=MySqlHook) -> Engine:
    return cli_hook(conn_id, hook_class).get_sqlalchemy_engine()