如何在Python中查询/修改数据库序列？（基于Alembic与SQLAlchemy）

因为SQLAlchemy的Metadata默认不会自动加载数据库中已有的序列，且Operations对象主要用于迁移操作，直接获取序列对象需要通过原生SQL查询或显式反射实现，以下是具体方案：

1. 直接执行原生SQL验证序列状态 #

不同数据库系统查询序列的SQL语句不同，直接通过已建立的连接执行对应SQL即可验证序列是否存在、获取当前值或修改序列：

示例（按数据库类型区分） #

from sqlalchemy import text
from sqlalchemy.exc import ProgrammingError

# 假设self.conn是你代码中已建立的数据库连接

# PostgreSQL: 查询序列是否存在
exists_result = self.conn.execute(
    text("SELECT EXISTS (SELECT 1 FROM information_schema.sequences WHERE sequence_name = :seq_name)"),
    {"seq_name": "test_sequence"}
).scalar()
assert exists_result is True  # 验证创建成功

# PostgreSQL: 获取序列当前值
current_val = self.conn.execute(
    text("SELECT currval('test_sequence')")
).scalar()

# PostgreSQL: 修改序列（例如重置起始值）
self.conn.execute(
    text("ALTER SEQUENCE test_sequence RESTART WITH 10")
)

# MySQL 8.0+: 查询序列是否存在
exists_result = self.conn.execute(
    text("SELECT EXISTS (SELECT 1 FROM information_schema.sequences WHERE sequence_name = :seq_name)"),
    {"seq_name": "test_sequence"}
).scalar()

# Oracle: 查询序列是否存在
exists_result = self.conn.execute(
    text("SELECT COUNT(*) FROM user_sequences WHERE sequence_name = 'TEST_SEQUENCE'")
).scalar()
assert exists_result == 1

2. 通过SQLAlchemy反射加载序列对象 #

使用MetaData.reflect()显式加载数据库中的序列，之后可从metadata.sequences集合中获取序列对象：

from sqlalchemy import MetaData

# 创建Metadata并绑定到你的engine
metadata = MetaData(bind=self.engine)
# 反射数据库中的所有对象，指定include_sequences=True以加载序列
metadata.reflect(include_sequences=True)

# 根据序列名称获取对象
target_sequence = metadata.sequences.get("test_sequence")
assert target_sequence is not None  # 验证序列存在

# 查看序列属性（例如起始值、增量）
assert target_sequence.start == 1
assert target_sequence.increment == 1

3. 从功能层面验证序列操作（单元测试推荐） #

无需直接获取序列对象，通过验证序列的功能来确认操作生效：

• 创建序列后，执行取值语句（如PostgreSQL的nextval），确认能正常获取值
• 删除序列后，再次执行取值语句会抛出异常，以此验证删除成功

# 验证创建成功：获取序列下一个值
next_val = self.conn.execute(text("SELECT nextval('test_sequence')")).scalar()
assert next_val == 1  # 匹配序列的起始值

# 调用你的drop_sequence方法
self.drop_sequence("test_sequence")

# 验证删除成功：执行取值会抛出异常
try:
    self.conn.execute(text("SELECT nextval('test_sequence')"))
    assert False, "序列未被删除"
except ProgrammingError:
    # 捕获数据库抛出的异常，确认删除生效
    pass

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