sqlalchemy-access库操作MS Access

因目前项目中数据处理的量稍大，为了方便和业务进行交互，对数据的加工和处理放到微软桌面数据库MS Access中。然后有些地方通过 Python 来操作 MS Access 数据库，用到 sqlalchemy-access库。本文对操作的要点做简单的描述。

之前写过一篇 Python 操作 MS Access 数据库不同的方法，有兴趣的小伙伴可以参考：Python如何操作MS Access数据库_python操作access数据库-CSDN博客。使用 sqlalchemy-access 的目的是为了简单。

odbc 数据源配置

在 Windows 下使用 ODBC ，是要配置 ODBC 数据源。设置的作用是允许通过名称来访问特定的数据库。在 command 窗口或者运行窗口输入 odbcad32 命令，然后点击「添加」按钮。

选择 MS Access 驱动：

将连接的数据源命名为 TEST_DB，并且选择需要连接的数据库：

安装 sqlalchemy-access 组件和相关的组件

使用 pip 方式进行安装如下三个组件。我在安装 sqlacodegen 的时候，sqlalchemy 被替换为 1.4 版。所以如果需要使用特定的 sqlalchemy 版本，可以在虚拟环境下操作。

pip install sqlalchemy
pip install sqlalchemy-access
pip install sqlacodegen

根据数据库表生成 model

因为数据库表已经存在，sqlalchemy 使用的时候，需要定义 Model。在这种情况下，通过 sqlacodegen 库来逆向生成 sqlalchemy 所需要的 model，在命令窗口执行如下命令：

sqlacodegen access+pyodbc://TEST_DB --outfile=test_db_models.py

参数说明：
第一个参数是访问数据库所需要的 url，sqlacodegen 和 sqlachem 使用相同的 URL
第二个参数：指定生成的文件

自动生成了 test_db_models.py 文件如下：

from sqlalchemy import Column, Float, String
from sqlalchemy_access.base import COUNTER
from sqlalchemy.ext.declarative import declarative_baseBase = declarative_base()
metadata = Base.metadataclass 北京(Base):__tablename__ = '北京'ID = Column(COUNTER, primary_key=True)专业类 = Column(String(255))专业代号 = Column(String(255))姓名 = Column(String(255))性别 = Column(String(255))称呼 = Column(String(255))原始分 = Column(Float)

对比数据库结构看一下：

CRUD 代码

在 basic_crud.py 文件中编写如下代码：


from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from test_db_models import 北京engine = create_engine("access+pyodbc://TEST_DB")
Session = sessionmaker(bind=engine)
session = Session()def test_query():result = session.query(北京).all()for row in result:print(row.ID, row.专业类, row.专业代号, row.姓名, row.性别, row.称呼, row.原始分)def test_query_by_name(name: str):result = session.query(北京).filter(北京.姓名 == name).first()print(result.ID, result.专业类, result.专业代号, result.姓名, result.性别, result.称呼, result.原始分)def test_insert():new_student = 北京(专业类='计算机科学与技术', 专业代号='001', 姓名='Alex', 性别='男', 称呼='先生', 原始分=590)session.add(new_student)session.commit()def test_update():student = session.query(北京).filter(北京.姓名 == 'Alex').first()student.原始分 = 600session.commit()def test_delete():student = session.query(北京).filter(北京.姓名 == 'Alex').first()session.delete(student)session.commit()