FastAPI微服务分区表pytest测试异常问题排查

问题根源 #

SQLAlchemy的Base.metadata.create_all会根据模型定义生成建表语句，如果子表模型继承了主分区表的分区配置（比如postgresql_partition_by），或者未明确标记为主分区表的继承子表，就会被错误创建为嵌套分区表，导致插入数据时找不到子分区而报错。

具体解决步骤 #

1. 修正子表模型定义 #

确保子表模型仅标记为继承主分区表，不设置自身的分区规则：

from sqlalchemy import Column, Integer, DateTime, PrimaryKeyConstraint
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

# 主分区表定义
class LoginHistory(Base):
    __tablename__ = "login_history"
    id = Column(Integer, nullable=False)
    created_at = Column(DateTime, nullable=False)
    # 其他业务字段...
    
    __table_args__ = (
        PrimaryKeyConstraint("id", "created_at"),  # 分区表主键必须包含分区键
        {"postgresql_partition_by": "RANGE (created_at)"}
    )

# 2024子表定义（普通表，继承主分区表）
class LoginHistoryY2024(Base):
    __tablename__ = "login_history_y2024"
    # 字段需与主表完全一致
    id = Column(Integer, nullable=False)
    created_at = Column(DateTime, nullable=False)
    # 其他业务字段...
    
    __table_args__ = (
        PrimaryKeyConstraint("id", "created_at"),
        {
            "postgresql_inherits": "login_history",
            "postgresql_partition_bound": "(RANGE ('2024-01-01'::DATE, '2025-01-01'::DATE))"
        }
    )

# 2025子表定义同理
class LoginHistoryY2025(Base):
    __tablename__ = "login_history_y2025"
    id = Column(Integer, nullable=False)
    created_at = Column(DateTime, nullable=False)
    # 其他业务字段...
    
    __table_args__ = (
        PrimaryKeyConstraint("id", "created_at"),
        {
            "postgresql_inherits": "login_history",
            "postgresql_partition_bound": "(RANGE ('2025-01-01'::DATE, '2026-01-01'::DATE))"
        }
    )

关键是子表不能设置postgresql_partition_by，仅通过postgresql_inherits关联主表，postgresql_partition_bound指定分区范围。

2. 测试环境使用Alembic迁移创建表 #

放弃用Base.metadata.create_all，直接复用生产环境的Alembic迁移脚本创建测试表，确保表结构完全一致：
在conftest.py中添加测试前置/后置操作：

import pytest
from alembic.config import Config
from alembic import command

@pytest.fixture(scope="session")
def setup_test_db():
    # 加载Alembic配置
    alembic_cfg = Config("alembic.ini")
    # 执行迁移到最新版本
    command.upgrade(alembic_cfg, "head")
    yield  # 测试执行阶段
    # 测试完成后回滚到初始状态
    command.downgrade(alembic_cfg, "base")

然后在测试用例中依赖这个fixture即可。

3. 检查Alembic迁移脚本正确性 #

确保迁移脚本中创建子表的语句是普通表+继承+分区边界，而非分区表：

def upgrade() -> None:
    # 创建主分区表
    op.create_table(
        "login_history",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("created_at", sa.DateTime(), nullable=False),
        # 其他字段
        sa.PrimaryKeyConstraint("id", "created_at"),
        postgresql_partition_by="RANGE (created_at)",
    )

    # 创建2024子表（普通表）
    op.create_table(
        "login_history_y2024",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("created_at", sa.DateTime(), nullable=False),
        # 其他字段与主表一致
        sa.PrimaryKeyConstraint("id", "created_at"),
        postgresql_inherits="login_history",
        postgresql_partition_bound="(RANGE ('2024-01-01'::DATE, '2025-01-01'::DATE))",
    )

    # 创建2025子表（普通表）
    op.create_table(
        "login_history_y2025",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("created_at", sa.DateTime(), nullable=False),
        # 其他字段与主表一致
        sa.PrimaryKeyConstraint("id", "created_at"),
        postgresql_inherits="login_history",
        postgresql_partition_bound="(RANGE ('2025-01-01'::DATE, '2026-01-01'::DATE))",
    )

验证 #

执行测试时，检查PostgreSQL中的表类型：

SELECT relname, relkind FROM pg_class WHERE relname LIKE 'login_history%';

主表login_history的relkind应为p（分区表），子表login_history_y2024、login_history_y2025的relkind应为r（普通表），此时插入数据即可正常命中子分区。

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