In MsSqlHook, SQLAlchemy engine scheme is overriden by the change in #40669

[emredjan]

Apache Airflow Provider(s)

common-sql

Versions of Apache Airflow Providers

apache-airflow-providers-common-sql==1.16.0
apache-airflow-providers-microsoft-mssql==3.9.0
apache-airflow-providers-odbc==4.7.0

Apache Airflow version

2.10.2

Operating System

RHEL 8.10

Deployment

Virtualenv installation

Deployment details

No response

What happened

We're using mssql+pyodbc connections in airflow to connect to our MS SQL Server. We're creating a hook using MsSqlHook and using .get_sqlalchemy_engine() to get an sql alchemy engine object to pass along our database tasks.
Prior to apache-airflow-providers-common-sql 1.15.0, this was working as expected (last known working version is 1.14.1, please see below for a working example)
With a change introduced in 1.15.0 (#40669), .get_sqlalchemy_engine() passes the creator argument with the value self.get_conn to the .create_engine() function, which overrides the engine creation with the default connection scheme, which uses pymssql, even though we specifically define the scheme as pyodbc in the connection, and in the hook.

Resulting engine object becomes a weird combination of two, with a mssql+pyodbc scheme in the connection uri, but tries to connect to using the pymssql dialect internally, which results in fatal errors:

As if it's trying to make a pymssql connection, but with a pyodbc URL.

What you think should happen instead

The connection should work, as it does prior to the change.

If we comment out / delete the line engine_kwargs["creator"] = self.get_conn in ‎airflow/providers/common/sql/hooks/sql.py (f6c7388#diff-6e1b2f961cb951d05d66d2d814ef5f6d8f8bf8f43c40fb5d40e27a031fed8dd7R246), connections works as expected.

How to reproduce

This is the connection that's used (values changed for privacy)

Basically use any MSSQL connection with a mssql+pyodbc scheme

{
    "conn_type": "mssql",
    "host": "<redacted>",
    "login": "user",
    "password": "pass",
    "schema": "master",
    "port": 1433,
    "extra": {
        "driver": "ODBC Driver 18 for SQL Server",
        "encrypt": "yes",
        "sqlalchemy_scheme": "mssql+pyodbc"
    }
}

Try to connect through a hook and sqlalchemy engine:

from airflow.providers.microsoft.mssql.hooks.mssql import MsSqlHook
hook = MsSqlHook(mssql_conn_id='conn1')  # conn1: the id for above connection
#hook = MsSqlHook(mssql_conn_id='conn1', sqlalchemy_scheme='mssql+pyodbc') # optional, same result
engine = hook.get_sqlalchemy_engine()
engine.connect()

This fails on > 1.15.0
Works on <= 1.14.1

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

[dabla]

Weird, because we are also using MSSQL with ODBC and are using the get_sqlalchemy_engine() method as well and having the same connection config as you posted without issues.

[emredjan]

Weird, because we are also using MSSQL with ODBC and are using the get_sqlalchemy_engine() method as well and having the same connection config as you posted without issues.

Are you using "sqlalchemy_scheme": "mssql+pyodbc" in the connection or the hook? Or you just use the ODBC Driver with the default SQLAlchemy scheme (mssql+pymssql)

[dabla]

This error is weird, but now I understand why, what you are doing is wrong, and in the past this wouldn't have caused any issues, but now it is.

Don't do this as you show above:

from airflow.providers.microsoft.mssql.hooks.mssql import MsSqlHook
hook = MsSqlHook(mssql_conn_id='conn1')  # conn1: the id for above connection
#hook = MsSqlHook(mssql_conn_id='conn1', sqlalchemy_scheme='mssql+pyodbc') # optional, same result
engine = hook.get_sqlalchemy_engine()
engine.connect()

But do this instead, as this will work, and is also better way of doing it:

hook = DbApiHook.get_hook(conn_id='conn1')
engine = hook.get_sqlalchemy_engine()
engine.connect()

I will explain why your example fails now. As you stated, you use MSSQL through ODBC connection (and thus ODBC driver), but in your example you use the MsSqlHook instead of OdbcHook, so here you try to us an ODBC connection with a PyMSSQL based hook (e.g. MsSqlHook), which of course won't work. It could have worked in the past but then you were lucky, now it won't as the get_sqlalchemy_engine method is now a generic method in DbApiHook which isn't overridden anymore in MsSqlHook.

Important:: It is always a good practice not to instantiate the specialized hook yourself, it's better to use the DbApiHook.get_hook(conn_id="conn_id") classmethod, as this will return you the specialized hook depending on the connection type of the connection id you passed. Depending of the connection type defined in the connection, Airflow will be smart enough to return you the correct Hook.

So please try my suggestion and let me know.

[dabla]

Weird, because we are also using MSSQL with ODBC and are using the get_sqlalchemy_engine() method as well and having the same connection config as you posted without issues.

Are you using "sqlalchemy_scheme": "mssql+pyodbc" in the connection or the hook? Or you just use the ODBC Driver with the default SQLAlchemy scheme (mssql+pymssql)

Yes, I also defined the "sqlalchemy_scheme": "mssql+pyodbc" in the extra of the connection as you showed above, the configuration of your connection is correct, but the instantiation of the hook isn't. Please check my answer above.

[emredjan]

Tried this:

from airflow.providers.common.sql.hooks.sql import DbApiHook
hook = DbApiHook.get_hook(conn_id='conn1')
engine = hook.get_sqlalchemy_engine()
engine.connect()

It results in the same error as above: AttributeError: 'pymssql._pymssql.Connection' object has no attribute 'add_output_converter' (This is an error from SQLAlchemy, when it internally tries to pass a pymssql connection object into a pyodbc dialect based connection script)

While I understand DbApiHook.get_hook() is the proper way to instantiate a database hook, it just creates a MsSqlHook with the same parameters as above. In practice, nothing changes on the engine creation side with .get_sqlalchemy_engine().

As I don't use any other databases, I'm not entirely sure about the benefits of passing the connection to the engine with the creator argument (engine_kwargs["creator"] = self.get_conn), but that causes any connection parameters to be ignored. As the passed connection with the get_conn just returns the default MS SQL connection with pymssql, it creates a basic connection, and any other scheme or other parameters in the connection are bypassed. See the get_conn definition:

def get_conn(self):
    """Return a connection object."""
    db = self.get_connection(self.get_conn_id())
    return self.connector.connect(host=db.host, port=db.port, username=db.login, schema=db.schema)

From SQLAlchemy create_engine() docs:

The resulting engine is a weird object with a pyodbc driver and dialect, but with a pymssql connection underlying.

[dabla]

Tried this:

from airflow.providers.common.sql.hooks.sql import DbApiHook
hook = DbApiHook.get_hook(conn_id='conn1')
engine = hook.get_sqlalchemy_engine()
engine.connect()

It results in the same error as above: AttributeError: 'pymssql._pymssql.Connection' object has no attribute 'add_output_converter' (This is an error from SQLAlchemy, when it internally tries to pass a pymssql connection object into a pyodbc dialect based connection script)

While I understand DbApiHook.get_hook() is the proper way to instantiate a database hook, it just creates a MsSqlHook with the same parameters as above. In practice, nothing changes on the engine creation side with .get_sqlalchemy_engine().

As I don't use any other databases, I'm not entirely sure about the benefits of passing the connection to the engine with the creator argument (engine_kwargs["creator"] = self.get_conn), but that causes any connection parameters to be ignored. As the passed connection with the get_conn just returns the default MS SQL connection with pymssql, it creates a basic connection, and any other scheme or other parameters in the connection are bypassed. See the get_conn definition:

def get_conn(self):
    """Return a connection object."""
    db = self.get_connection(self.get_conn_id())
    return self.connector.connect(host=db.host, port=db.port, username=db.login, schema=db.schema)

From SQLAlchemy create_engine() docs:

The resulting engine is a weird object with a pyodbc driver and dialect, but with a pymssql connection underlying.

Ok now I'm confused... Are you trying to access MSSQL through ODBC or PyMSQL? Because in your initial post you stated ODBC, but now you're referring to PyMSSQL? If you want to connect through PDBC, make sure the connection type is ODBC and not MSSQL which will use pymssql underneath.

[dabla]

This config is also contradicting:

{
    "conn_type": "mssql",
    "host": "<redacted>",
    "login": "user",
    "password": "pass",
    "schema": "master",
    "port": 1433,
    "extra": {
        "driver": "ODBC Driver 18 for SQL Server",
        "encrypt": "yes",
        "sqlalchemy_scheme": "mssql+pyodbc"
    }
}

Connection type is mssql (so pymssql), but in your extras you specify you want to use the ODBC driver and "mssql+pyodbc" sqlalchemy_scheme while the connection type is mssql which is actually pymssql, not ODBC. Try changing the connection type to ODBC.