SQLAlchemy dynamic construction of GENERATED column

Question

I want to dynamically derive columns with the server_default set to Computed(...).

The statement going into Computed(...) has to be derived from other columns, and therefore needs to have access to the class after the relationships have been mapped:

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
"""

Where the query is constructed as such:

-- 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

I know how to construct the query from the relationships, but I don’t know how to tell SQLAlchemy how to map everything except for my properties, then finish mapping while allowing those properties access to the class’s relationships.

The answer here is NOT:

• “Simply use a column_property”: Those accept only finished columns, and don’t allow to construct them with access to the mapped class
• “Use sqlalchemy_utils.aggregate”: That one is implemented via after_flush trigger, not GENERATED ALWAYS AS.

Answer 1

It is indeed relatively easily possible, by creating a subclass of Computed that can be passed to the column. That subclass defines sqltext as a property instead of a instance variable. Since that property is only accessed after mapping, it has access to the fully mapped model class.

Unfortunately what I actually wanted to do with it is impossible, as the generated statement is not valid in PostgreSQL (GENERATED ALWAYS AS cannot use nested queries).

Nevertheless the approach might be useful for other dynamically generated GENERATED ALWAYS AS statements.

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()