SQLAlchemy

Question

我有两个 table。一个被映射为 ORM class，如下所示：

from typing import Union
from dataclasses import dataclass
from api.models import db, SerializerMixin

@dataclass
class Company(db.Model, SerializerMixin):
    __tablename__ = "Company"
    company_id: int = db.Column(db.Integer, primary_key = True)
    name: str = db.Column(db.String(50), nullable = False, unique = True)
    primary_contact: Union[int, None] = db.Column(db.Integer, db.ForeignKey("User.user_id"))
    website: Union[str, None] = db.Column(db.Text)

第二个写成普通的SQL table（这样写是因为我不打算直接从这个table查询，只用于JOINS和计数）：

user_company_table = db.Table("UserCompany",
    db.Column("user_id", db.Integer, db.ForeignKey("User.user_id"), primary_key = True, unique = True),
    db.Column("company_id", db.Integer, db.ForeignKey("Company.company_id"), primary_key = True),
    db.Column("assigned", db.DateTime, server_default = text("CURRENT_TIMESTAMP"), onupdate = datetime.utcnow),
    db.Column("approved", db.Boolean, nullable = False, default = False)
)

我需要从 Company 获取所有列，同时获取拥有 approved = true 并分配给一个特定公司的所有用户的计数。结果 SQL 看起来像这样：

SELECT Company.company_id, Company.name, Company.primary_contact, Company.website, (SELECT COUNT(1) FROM UserCompany WHERE approved = true and company_id = Company.company_id) AS user_count FROM Company;

这应该给我这个示例结果：

所以 user_count 是附加到公司 table 的子查询的结果集。我将如何向 Company class 添加一个 user_count 属性以便在我运行 Company.query.all() 时急切加载此子查询并将结果作为列附加？

备注：使用 Flask 2.0+、Flask-SQLAlchemy、Python 3.9（解释模型属性中的@dataclass 和类型提示）

Answer 1

我认为你可以为此使用 correlated subquery relationship hybrid:

class Company:
  ...

  @hybrid_property
  def user_count(self):
    return db.session.query(user_company_table)\
      .filter_by(company_id=self.id, approved=True)\
      .count()

  @user_count.expression
  def user_count(cls):
    return select(func.count(1))\
      .where(user_company_table.c.company_id==cls.id)\
      .where(user_company_table.c.approved==True)\
      .label('user_count')

您甚至可以选择带有标量子查询的非常短的 column_property：https://docs.sqlalchemy.org/en/14/orm/mapped_sql_expr.html#using-column-property

SQLAlchemy - 预加载子查询作为 ORM 的列 class

SQLAlchemy - Eager load subquery as column for an ORM class

python

mysql

flask

flask-sqlalchemy