时间戳

首页 > 解决方案 > 生成列的 SQLAlchemy 动态构造

python - 生成列的 SQLAlchemy 动态构造

问题描述

我想动态派生server_default设置为Computed(...).

进入的语句Computed(...)必须从其他列派生，因此需要在映射s后访问该类：relationship

from uuid import uuid4
from sqlalchemy import func, Computed, ForeignKey, Column
from sqlalchemy.orm import relationship
from sqlalchemy.sql.base import Executable
from sqlalchemy.ext.declarative import as_declarative
from sqlalchemy.dialects.postgresql import INTEGER, UUID

def aggregate_computed(relationship: str, column: Column, expr: Executable):
    ...

@as_declarative()
class Base: pass

class Other(Base):
    __tablename__ = "other"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
    this = relationship("This", back_populates="other")
    this_id = Column(UUID(as_uuid=True), ForeignKey("this.id"))

class This(Base, IDMixin):
    __tablename__ = "this"
    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid4)
    other = relationship("Other", back_populates="this")
    other_count = aggregate_computed("species", Column(INTEGER), func.count(1))

assert isinstance(This.other_count.server_default, Computed)
assert This.other_count.server_default.sqltext == """\
SELECT count(1) AS out
FROM other
WHERE this.id = other.this_id
"""

查询构造如下：

-- this is the decorated function
SELECT count(1) AS out
-- this is derived from the relationships.
-- for a m-to-n relation, it would make use of the `secondary` table
FROM other WHERE this.id = other.this_id

我知道如何从关系中构造查询，但我不知道如何告诉 SQLAlchemy 如何映射除我的属性之外的所有内容，然后在允许这些属性访问类的关系的同时完成映射。

这里的答案不是：

标签： pythonpostgresqlsqlalchemygenerated-columns

解决方案

Computed通过创建可以传递给列的子类，这确实相对容易。该子类定义sqltext为属性而不是实例变量。由于该属性仅在映射后才能访问，因此它可以访问完全映射的模型类。

不幸的是，我真正想做的事情是不可能的，因为生成的语句在 PostgreSQL 中无效（GENERATED ALWAYS AS不能使用嵌套查询）。

然而，该方法可能对其他动态生成的GENERATED ALWAYS AS语句很有用。

class ComputedAggregate(Computed):
    def __init__(self, relationship: str, expr, *, persisted=None):
        # do not call super().__init__, as it tries to set self.sqltext
        self.relationship = relationship
        self.expr = expr
        self.persisted = persisted
        self.column = None

    @property
    def sqltext(self):
        cls = get_class_by_table(MyBase, self.column.table)
        relationships: t.List[RelationshipProperty] = list(
            reversed(path_to_relationships(self.relationship, cls))
        )
        query = select_correlated_expression(
            cls,
            self.expr,
            self.relationship,
            relationships[0].mapper.class_,
        )
        return query.as_scalar()

推荐阅读