Postgresql分区与SQLAlchemy

Question

SQLAlchemy文档解释如何创建分区表。但它并没有解释如何创建分区。

所以，如果我有这个：

#Skipping create_engine and metadata
Base = declarative_base()

class Measure(Base):
    __tablename__ = 'measures'
    __table_args__ = {
        postgresql_partition_by: 'RANGE (log_date)'
    }
    city_id = Column(Integer, not_null=True)
    log_date = Columne(Date, not_null=True)
    peaktemp = Column(Integer)
    unitsales = Column(Integer)

class Measure2020(Base):
    """How am I suppposed to declare this ? """

我知道我将会执行大部分的操作是 SELECT * FROM measures WHERE logdate between XX and YY。但这似乎很有趣。

- Rémi Desgrange

---

我认为在这种情况下，您最好的选择可能是使用原始SQL。 - Ilja Everilä

看到其他人在解决这个问题，感觉很有成就感。 - Adam Bethke

Mike Bayer在这里给出了答案：https://github.com/sqlalchemy/sqlalchemy/issues/5313 - Rémi Desgrange

Answer 1

你可以使用一个 MeasureMixin，这两个类都可以继承它。 然后使用一个 event 来附加表分区。

from sqlalchemy import event

class MeasureMixin:
    city_id = Column(Integer, not_null=True)
    log_date = Column(Date, not_null=True)
    peaktemp = Column(Integer)
    unitsales = Column(Integer)

class Measure(MeasureMixin, Base):
    __tablename__ = 'measures'
    __table_args__ = {
        postgresql_partition_by: 'RANGE (log_date)'
    }

class Measure2020(MeasureMixin, Base):
    __tablename__ = 'measures2020'

Measure2020.__table__.add_is_dependent_on(Measure.__table__)

event.listen(
    Measure2020.__table__,
    "after_create",
    DDL("""ALTER TABLE measures ATTACH PARTITION measures2020
VALUES FROM ('2020-01-01') TO ('2021-01-01');""")
)

- moshevi

---

很好，甚至可以自动创建分区。 - Rémi Desgrange

我应该在Measure2020和Measure之间添加依赖关系。将进行编辑。 - moshevi

嗯，postgresql_partition_by 周围不应该加引号吗？ - Roman Nakutnyi

Answer 2

我遇到了类似的问题。我发现@moshevi的答案非常有用，最终我对其进行了一些泛化（因为我需要对多个表进行分区）。

首先创建一个这样的元类：

from sqlalchemy.ext.declarative import DeclarativeMeta
from sqlalchemy.sql.ddl import DDL
from sqlalchemy import event


class PartitionByYearMeta(DeclarativeMeta):
    def __new__(cls, clsname, bases, attrs, *, partition_by):
        @classmethod
        def get_partition_name(cls_, key):
            # 'measures' -> 'measures_2020' (customise as needed)
            return f'{cls_.__tablename__}_{key}'
        
        @classmethod
        def create_partition(cls_, key):
            if key not in cls_.partitions:
                
                Partition = type(
                    f'{clsname}{key}', # Class name, only used internally
                    bases,
                    {'__tablename__': cls_.get_partition_name(key)}
                )
                
                Partition.__table__.add_is_dependent_on(cls_.__table__)
                
                event.listen(
                    Partition.__table__,
                    'after_create',
                    DDL(
                        # For non-year ranges, modify the FROM and TO below
                        f"""
                        ALTER TABLE {cls_.__tablename__}
                        ATTACH PARTITION {Partition.__tablename__}
                        FOR VALUES FROM ('{key}-01-01') TO ('{key+1}-01-01');
                        """
                    )
                )
                
                cls_.partitions[key] = Partition
            
            return cls_.partitions[key]
        
        attrs.update(
            {
                # For non-RANGE partitions, modify the `postgresql_partition_by` key below
                '__table_args__': attrs.get('__table_args__', ())
                + (dict(postgresql_partition_by=f'RANGE({partition_by})'),),
                'partitions': {},
                'partitioned_by': partition_by,
                'get_partition_name': get_partition_name,
                'create_partition': create_partition
            }
        )
        
        return super().__new__(cls, clsname, bases, attrs)

下一步，对于您模型中想要进行分区的任何表格：

class MeasureMixin:
    # The columns need to be pulled out into this mixin
    # Note: any foreign key columns will need to be wrapped like this:

    @declared_attr
    def city_id(self):
        return Column(ForeignKey('cities.id'), not_null=True)
    
    log_date = Column(Date, not_null=True)
    peaktemp = Column(Integer)
    unitsales = Column(Integer)

class Measure(MeasureMixin, Base, metaclass=PartitionByYearMeta, partition_by='logdate'):
    __tablename__ = 'measures'

这使得添加更多表格和按任意数量的值进行分区变得容易。

即兴创建新的分区如下所示：

# Make sure you commit any session that is currently open, even for select queries:
session.commit()

Partition = Measure.create_partition(2020)
if not engine.dialect.has_table(Partition.__table__.name):
    Partition.__table__.create(bind=engine)

现在已经为关键字2020创建了分区，可以插入该年份的值。

- Seb

---

@moshevi，我正在尝试扩展您的代码。如果您能回答这个[https://stackoverflow.com/questions/66562240/postgresql-partition-management-with-sqlalchemy-and-flask-migrate] SO问题，那将非常有帮助。 - DGL

Answer 3

也许有点晚了，但我想分享一下我基于@moshevi和@Seb的答案所构建的东西：

在我的物联网应用场景中，我需要实际的子分区（第一级为“year”，第二级为“nodeid”）。此外，我还希望稍微进行一些泛化。

这就是我想出来的内容：

from sqlalchemy.ext.declarative import DeclarativeMeta
from sqlalchemy.sql.ddl import DDL
from sqlalchemy import event

class PartitionByMeta(DeclarativeMeta):
    def __new__(cls, clsname, bases, attrs, *, partition_by, partition_type):

        @classmethod
        def get_partition_name(cls_, suffix):
            return f'{cls_.__tablename__}_{suffix}'

        @classmethod
        def create_partition(cls_, suffix, partition_stmt, subpartition_by=None, subpartition_type=None):
            if suffix not in cls_.partitions:

                partition = PartitionByMeta(
                    f'{clsname}{suffix}',
                    bases,
                    {'__tablename__': cls_.get_partition_name(suffix)},
                    partition_type = subpartition_type,
                    partition_by=subpartition_by,
                )

                partition.__table__.add_is_dependent_on(cls_.__table__)

                event.listen(
                    partition.__table__,
                    'after_create',
                    DDL(
                        # For non-year ranges, modify the FROM and TO below
                        # LIST: IN ('first', 'second');
                        # RANGE: FROM ('{key}-01-01') TO ('{key+1}-01-01')
                        f"""
                        ALTER TABLE {cls_.__tablename__}
                        ATTACH PARTITION {partition.__tablename__}
                        {partition_stmt};
                        """
                    )
                )
                
                cls_.partitions[suffix] = partition
            
            return cls_.partitions[suffix]
        
        if partition_by is not None:
            attrs.update(
                {
                    '__table_args__': attrs.get('__table_args__', ())
                    + (dict(postgresql_partition_by=f'{partition_type.upper()}({partition_by})'),),
                    'partitions': {},
                    'partitioned_by': partition_by,
                    'get_partition_name': get_partition_name,
                    'create_partition': create_partition
                }
            )
        
        return super().__new__(cls, clsname, bases, attrs)

假设由 @moshevi 引入的 VehicleDataMixin 类已被创建，下面介绍如何使用：

class VehicleData(VehicleDataMixin, Project, metaclass=PartitionByMeta, partition_by='timestamp',partition_type='RANGE'):
    __tablename__ = 'vehicle_data'
    __table_args__ = (
        Index('ts_ch_nod_idx', "timestamp", "nodeid", "channelid", postgresql_using='brin'),
        UniqueConstraint('timestamp','nodeid','channelid', name='ts_ch_nod_constr')
    )

然后可以像这样迭代地进行子分区（需进行适当调整）

    for y in range(2017, 2021): 
         # Creating tables for all known nodeids
        tbl_vehid_y = VehicleData.create_partition(
            f"{y}", partition_stmt=f"""FOR VALUES FROM ('{y}-01-01') TO ('{y+1}-01-01')""",
            subpartition_by='nodeid', subpartition_type='LIST'
        )
        
        for i in {3, 4, 7, 9}:
            # Creating all the years below these nodeids including a default partition
            tbl_vehid_y.create_partition(
                f"nid{i}", partition_stmt=f"""FOR VALUES IN ('{i}')"""
            )
        
        # Defaults (nodeid) per year partition
        tbl_vehid_y.create_partition("def", partition_stmt="DEFAULT")

   # Default to any other year than anticipated
   VehicleData.create_partition("def", partition_stmt="DEFAULT")

partition_by='timestamp' <= 这是用于分区的列

partition_type='RANGE' <= 这是（特定于PSQL）的分区类型

partition_stmt=f"""FOR VALUES IN ('{i}')""" <= 这是（特定于PSQL）的分区语句。

- jenszo

Answer 4

对于数据库分区（PostgreSQL或MySQL），您可以尝试使用architect包。它与一系列ORM库兼容，包括SQLAlchemy。以下是PostgreSQL的示例- https://architect.readthedocs.io/features/partition/postgresql.html 。支持多种分区类型，希望能满足您的要求。

- Tsvetan Kintisheff