Also use ODBC connection for sqlalchemy engine in OdbcHook like JdbcHook

providers/src/airflow/providers/odbc/hooks/odbc.py

@@ -180,6 +180,19 @@ def connect_kwargs(self) -> dict:
 
         return merged_connect_kwargs
 
+    def get_sqlalchemy_engine(self, engine_kwargs=None):
+        """
+        Get an sqlalchemy_engine object.
+
+        :param engine_kwargs: Kwargs used in :func:`~sqlalchemy.create_engine`.
+        :return: the created engine.
+        """
+        if engine_kwargs is None:
+            engine_kwargs = {}
+        engine_kwargs["creator"] = self.get_conn
+
+        return super().get_sqlalchemy_engine(engine_kwargs)
+
     def get_conn(self) -> Connection:
         """Return ``pyodbc`` connection object."""
         conn = connect(self.odbc_connection_string, **self.connect_kwargs)

providers/tests/odbc/hooks/test_odbc.py

@@ -19,6 +19,7 @@
 
 import json
 import logging
+import sqlite3
 from dataclasses import dataclass
 from unittest import mock
 from unittest.mock import patch
@@ -340,3 +341,11 @@ def test_query_no_handler_return_none(self):
         hook = mock_hook(OdbcHook)
         result = hook.run("SQL")
         assert result is None
+
+    def test_get_sqlalchemy_engine_verify_creator_is_being_used(self):
+        hook = mock_hook(OdbcHook, conn_params={"extra": {"sqlalchemy_scheme": "sqlite"}})
+
+        with sqlite3.connect(":memory:") as connection:
+            hook.get_conn = lambda: connection
+            engine = hook.get_sqlalchemy_engine()
+            assert engine.connect().connection.connection == connection