如何用SQLAlchemy定义PostgreSQL列表分区表及具体分区？

环境信息 #

• psycopg2(2.9.9)
• SQLAlchemy(2.0.22)
• PostgreSQL(16.0)

解决方案 #

针对你的需求，以下两种方式可以在SQLAlchemy中同时定义主分区表和具体分区：

方式一：声明式模型结合DDL事件监听 #

保留你原有的声明式模型结构，通过监听主表创建事件，自动执行分区创建的SQL语句：

from sqlalchemy import create_engine, Column, Integer, String, DDL, event
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine('postgresql://test_user:test_password@localhost:5432/test_db')
Base = declarative_base()

class PartitionedTable(Base):
    __tablename__ = 'partitioned_table'
    __table_args__ = {
        'postgresql_partition_by': 'LIST (category)',
    }

    id = Column(Integer, primary_key=True)
    name = Column(String)
    category = Column(String)
    value = Column(Integer)

# 定义分区A的创建语句
create_partition_a = DDL("""
    CREATE TABLE partitioned_table_a PARTITION OF partitioned_table
    FOR VALUES IN ('A');
""")

# 定义分区B的创建语句
create_partition_b = DDL("""
    CREATE TABLE partitioned_table_b PARTITION OF partitioned_table
    FOR VALUES IN ('B');
""")

# 绑定事件：主表创建完成后自动创建分区
event.listen(PartitionedTable.__table__, 'after_create', create_partition_a)
event.listen(PartitionedTable.__table__, 'after_create', create_partition_b)

# 执行所有表的创建（主表+分区）
Base.metadata.create_all(engine)

方式二：使用SQLAlchemy Table对象显式定义分区 #

完全通过SQLAlchemy API配置分区，避免直接编写原生SQL：

from sqlalchemy import create_engine, Column, Integer, String, Table, MetaData
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine('postgresql://test_user:test_password@localhost:5432/test_db')
Base = declarative_base()
metadata = Base.metadata

# 定义主分区表
partitioned_table = Table(
    'partitioned_table', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String),
    Column('category', String),
    Column('value', Integer),
    postgresql_partition_by='LIST (category)'
)

# 定义分区A
partition_a = Table(
    'partitioned_table_a', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String),
    Column('category', String),
    Column('value', Integer),
    postgresql_partition_of=partitioned_table,
    postgresql_partition_for="VALUES IN ('A')"
)

# 定义分区B
partition_b = Table(
    'partitioned_table_b', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String),
    Column('category', String),
    Column('value', Integer),
    postgresql_partition_of=partitioned_table,
    postgresql_partition_for="VALUES IN ('B')"
)

# 创建所有表
metadata.create_all(engine)

关键说明 #

• 分区表的结构必须与主表完全一致，包括字段类型、主键和约束；
• 执行metadata.create_all()时，SQLAlchemy会自动按主表→分区的顺序创建，无需手动控制；
• 方式一更适合保留ORM模型风格，方式二则完全通过API配置，避免原生SQL依赖。

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