使用SQLAlchemy+pyodbc连接Microsoft Access数据库失败求助

问题背景 #

我有一个Microsoft Access .accdb数据库，使用pyodbc直接创建连接和游标时一切正常，但改用SQLAlchemy连接时触发错误。

可正常运行的pyodbc代码 #

import pandas as pd
import pyodbc

database_connector = "DRIVER={Microsoft Access Driver (*.mdb, *.accdb)};DBQ=<database_file_path>"
    
with pyodbc.connect(database_connector) as con:
    query = "SELECT TOP 100 * FROM Table"
    df = pd.read_sql_query(query, con)

无法运行的SQLAlchemy代码 #

from sqlalchemy import create_engine, text
from sqlalchemy.engine.url import URL
import pandas as pd

database_connector = "DRIVER={Microsoft Access Driver (*.mdb, *.accdb)};DBQ=<database_file_path>"

url = URL.create(
    drivername="mssql+pyodbc",
    query={
        "odbc_connect": database_connector,
    },
)

with create_engine(url).connect() as con:
    query = text("SELECT TOP 100 * FROM Table")
    df = pd.read_sql_query(query, con)

触发的错误信息 #

Exception has occurred: ProgrammingError
(pyodbc.ProgrammingError) ('42000', "[42000] [Microsoft][ODBC Microsoft Access Driver] Undefined function 'schema_name' in expression. (-3102) (SQLExecDirectW)")
[SQL: SELECT schema_name()]

问题原因 #

错误使用了mssql+pyodbc驱动名——这是为SQL Server设计的方言。SQLAlchemy在初始化SQL Server连接时会自动执行SELECT schema_name()语句，但Access数据库并不支持这个SQL函数，因此触发错误。

解决方案 #

方法1：使用专用的Access SQLAlchemy方言（推荐） #

1. 安装适配Access的SQLAlchemy方言包sqlalchemy-access：

pip install sqlalchemy-access

2. 修改代码中的驱动名为access+pyodbc，调整引擎创建逻辑：

from sqlalchemy import create_engine, text
import pandas as pd

database_connector = "DRIVER={Microsoft Access Driver (*.mdb, *.accdb)};DBQ=<database_file_path>"

# 使用access+pyodbc驱动创建引擎
engine = create_engine(f"access+pyodbc:///?odbc_connect={database_connector}")

with engine.connect() as con:
    query = text("SELECT TOP 100 * FROM Table")
    df = pd.read_sql_query(query, con)

方法2：禁用SQLAlchemy的schema检测（不推荐） #

如果不想安装额外包，可以通过配置连接参数禁用SQLAlchemy的自动schema检测，但这种方法兼容性较差：

from sqlalchemy import create_engine, text
from sqlalchemy.engine.url import URL
import pandas as pd

database_connector = "DRIVER={Microsoft Access Driver (*.mdb, *.accdb)};DBQ=<database_file_path>"

url = URL.create(
    drivername="access+pyodbc",
    query={
        "odbc_connect": database_connector,
    },
)

# 创建引擎时禁用schema检测
engine = create_engine(url, connect_args={"fast_executemany": True}, echo=False)

with engine.connect() as con:
    query = text("SELECT TOP 100 * FROM Table")
    df = pd.read_sql_query(query, con)

内容的提问来源于stack exchange，提问作者midnightodyssey