SQLAlchemy迁移MySQL至金仓数据库指南

从一个报错开始

最近接手一个任务，需要将一个基于SQLAlchemy的数据分析脚本从MySQL迁移到金仓数据库。本以为只是换个数据库驱动的小事，没想到一运行就迎面撞上一个错误：No module named 'sqlalchemy.dialects.kingbase'。

排查一番才发现，SQLAlchemy并不像对待MySQL、PostgreSQL那样自带金仓的驱动，它的方言包需要手动安装。整个过程折腾了近两天才跑通，这里把踩坑和解决的过程记录下来，供有类似需求的同行参考。

一、方言包是什么，为什么非要手动装

可以把SQLAlchemy理解为一个通用的“外壳”，它本身并不知道如何与具体的数据库通信。每种数据库都需要一个专门的“翻译官”，这个翻译官就是方言包。

对于MySQL、PostgreSQL这类主流数据库，SQLAlchemy在安装时就已经集成了它们的方言包。但金仓目前不在这个默认列表里，因此需要开发者自行从官网下载，并手动放置到指定目录。

另外需要注意的是，这个方言包的底层依赖于ksycopg2驱动，所以这两个组件都需要安装。

二、安装步骤（踩坑记录）

2.1 先装SQLAlchemy

pip install sqlalchemy

安装完成后，需要确认一下它的安装位置，后续放置方言包时会用到：

pip show sqlalchemy

以我的环境为例，输出信息如下：

Name: SQLAlchemy
Version: 1.4.36
Location: /usr/local/lib/python3.8/site-packages

请记下这个Location路径。

2.2 装ksycopg2驱动

pip install ksycopg2

这一步通常比较顺利。Windows用户可能需要安装对应的ksycopg2-win64版本。

2.3 放方言包（这里卡了我半天）

找到上一步记录的SQLAlchemy安装目录，进入其中的dialects文件夹：

cd /usr/local/lib/python3.8/site-packages/sqlalchemy/dialects

将从官网下载的kingbase文件夹整个复制到这个目录下。最终的目录结构应该是这样的：

sqlalchemy/dialects/
├── kingbase/
│   ├── __init__.py
│   └── base.py
├── mysql/
├── postgresql/
└── ...

这里有个常见的坑：最初我把kingbase文件夹直接放在了site-packages根目录下，导致SQLAlchemy始终无法加载它。后来仔细阅读文档才确认，必须放在dialects子目录下。

版本匹配问题：官方通常会提供适配不同SQLAlchemy主版本的方言包，例如1.3、1.4、2.0等。我使用的SQLAlchemy版本是1.4.36，因此选择了1.4版本的方言包。版本不匹配可能会导致一些奇怪的错误，例如提示找不到某个模块。

三、连接数据库

3.1 连接字符串怎么写

完成安装配置后，就可以在代码中建立连接了。连接字符串的格式如下：

from sqlalchemy import create_engine
engine = create_engine('kingbase+ksycopg2://SYSTEM:123456@127.0.0.1:54321/TEST')

简单拆解一下：

• kingbase：指定方言，告诉SQLAlchemy使用金仓的方言包。
• ksycopg2：指定底层驱动。
• SYSTEM：数据库用户名。
• 123456：对应用户的密码。
• 127.0.0.1：数据库服务器IP地址。
• 54321：端口号（金仓数据库默认端口）。
• TEST：要连接的具体数据库名。

实际上，+ksycopg2部分可以省略，直接写成kingbase://...即可，SQLAlchemy会默认使用ksycopg2作为驱动。

3.2 测试一下能不能连上

from sqlalchemy import create_engine
conn_str = 'kingbase://SYSTEM:123456@127.0.0.1:54321/TEST'
engine = create_engine(conn_str)
conn = engine.connect()
result = conn.execute("SELECT version()")
print(result.fetchone())
conn.close()

如果成功打印出数据库版本信息，那么恭喜，连接配置正确。

3.3 连接池参数（生产环境有用）

如果是开发Web服务或需要长期运行的脚本，建议配置连接池以优化性能和管理连接：

engine = create_engine(
    'kingbase://SYSTEM:123456@127.0.0.1:54321/TEST',
    pool_size=10,          # 连接池中保持的连接数
    max_overflow=20,       # 连接池耗尽时允许额外创建的最大连接数
    pool_recycle=3600,     # 连接使用多久后回收（单位：秒）
    pool_pre_ping=True     # 每次从池中取出连接前先进行健康检查
)

其中pool_pre_ping=True这个参数非常实用，它能有效避免应用程序拿到一个已经失效的数据库连接。

四、ORM建模和基本操作

连接建立后，就可以使用SQLAlchemy强大的ORM功能来操作数据库了。

4.1 定义模型

首先创建一个所有模型类的基类，然后定义具体的表映射类：

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, DateTime
from datetime import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = 'test_user'  # 指定对应的数据库表名
    
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100))
    created_at = Column(DateTime, default=datetime.now)
    
    def __repr__(self):
        return f"User(id={self.id}, name='{self.name}', email='{self.email}')"

__tablename__属性是必须的，它告诉ORM这个类对应数据库中的哪张表。

4.2 建表

# 自动创建所有定义的表（如果表不存在）
Base.metadata.create_all(engine)

这个操作是幂等的，多次执行不会导致重复建表。

4.3 创建Session

Session是ORM进行数据库操作的主要接口：

from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
session = Session()

4.4 增删改查

插入数据：

user1 = User(name='张三', email='zhangsan@test.com')
user2 = User(name='李四', email='lisi@test.com')
session.add(user1)
session.add(user2)
# 或者一次性添加多个对象：
session.add_all([user1, user2])
session.commit()

查询数据：

# 查询所有记录
users = session.query(User).all()
for u in users:
    print(u.name, u.email)

# 条件查询（获取第一条）
user = session.query(User).filter(User.name == '张三').first()
print(user.email)

# 模糊查询
users = session.query(User).filter(User.name.like('%张%')).all()

更新数据：

# 方式1：先查询出对象，修改属性后提交
user = session.query(User).filter(User.name == '张三').first()
user.email = 'newemail@test.com'
session.commit()

# 方式2：批量更新（直接使用update语句）
session.query(User).filter(User.name == '张三').update(
    {"email": "batch_update@test.com"})
session.commit()

删除数据：

# 先查询，再删除对象
user = session.query(User).filter(User.name == '张三').first()
session.delete(user)
session.commit()

# 批量删除（直接使用delete语句）
session.query(User).filter(User.id > 100).delete()
session.commit()

4.5 事务处理

Session默认工作在事务模式下，需要显式调用commit()更改才会持久化到数据库。如果操作过程中发生异常，可以进行回滚：

try:
    session.add(User(name='王五', email='wangwu@test.com'))
    # 此处如果发生错误...
    session.commit()
except Exception as e:
    session.rollback()
    print(f"操作失败: {e}")
finally:
    session.close()

4.6 完整跑一遍

将上述代码片段整合起来，形成一个完整的示例：

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Column, Integer, String

# 1. 连接数据库
engine = create_engine('kingbase://SYSTEM:123456@127.0.0.1:54321/TEST')

# 2. 定义模型
Base = declarative_base()
class User(Base):
    __tablename__ = 'test_user'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100))
    def __repr__(self):
        return f"User(id={self.id}, name={self.name})"

# 3. 创建表
Base.metadata.create_all(engine)

# 4. 使用Session进行数据操作
Session = sessionmaker(bind=engine)
session = Session()

# 插入数据
session.add_all([
    User(name='张三', email='zhangsan@test.com'),
    User(name='李四', email='lisi@test.com'),
])
session.commit()

# 查询并打印
for user in session.query(User).all():
    print(user)

session.close()

五、踩过的几个坑

坑1：方言包放错位置

这是最容易出错的地方。方言包必须严格放置在sqlalchemy/dialects/kingbase/目录下，直接扔到site-packages根目录是无效的。

坑2：版本不对应

SQLAlchemy 1.4.x 版本必须使用适配1.4的方言包。如果误用了1.3版本的包，通常会报“模块找不到”之类的错误。

坑3：libkci.so找不到

错误信息类似于 libkci.so: cannot open shared object file。解决方法是通过环境变量指定金仓库文件的路径：

export LD_LIBRARY_PATH=/home/kingbase/lib:$LD_LIBRARY_PATH

坑4：连接字符串写错

注意连接字符串的开头是kingbase://，不要误写成kingbase+psycopg2://。虽然ksycopg2是基于psycopg2修改的，但驱动名称写错会导致连接失败。

六、小结

整体梳理下来，使用SQLAlchemy连接金仓数据库的过程并不复杂，核心区别在于方言包需要手动配置。

主要流程可以归纳为三步：

1. 安装SQLAlchemy和ksycopg2驱动。
2. 将金仓方言包正确放置到sqlalchemy/dialects/目录下。
3. 使用kingbase://格式的连接字符串创建引擎。

一旦连接建立，后续的ORM操作与其他数据库完全一致，迁移成本主要集中在初始配置环节。如果项目本身已经采用SQLAlchemy作为ORM框架，那么切换数据库的代码改动量其实非常小。