Flask-SQLAlchemy为Company ORM类添加预加载user_count属性的实现方法
实现方案
首先确保关联表user_company_table的定义顺序在Company类之前,再导入需要的SQLAlchemy工具:
from sqlalchemy import func, select # 你的原有user_company_table关联表代码保持不变 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类,添加column_property类型的user_count属性即可实现查询时自动加载统计值:
@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) # 新增统计字段,查询时自动作为附加字段返回 user_count: int = db.column_property( select(func.count(user_company_table.c.user_id)) .where( user_company_table.c.company_id == company_id, user_company_table.c.approved == True ) .scalar_subquery() )
效果验证
直接调用Company.query.all()查询时,生成的SQL和你给出的目标SQL完全一致,每个返回的Company实例都会自带user_count属性,值为当前公司已通过审批的关联用户总数。因为你使用了@dataclass和SerializerMixin,接口序列化时也会自动包含该字段。
可选性能优化
如果大部分业务场景不需要统计该字段,可以改为延迟加载,避免每次查询都执行子查询损耗性能:
from sqlalchemy.orm import deferred # 改为延迟加载的写法 user_count: int = deferred( db.column_property( select(func.count(user_company_table.c.user_id)) .where( user_company_table.c.company_id == company_id, user_company_table.c.approved == True ) .scalar_subquery() ) )
需要加载该字段时主动指定查询选项即可:
from sqlalchemy.orm import undefer companies = Company.query.options(undefer(Company.user_count)).all()
内容的提问来源于stack exchange,提问作者OzzyTheGiant
相关产品推荐
相关产品推荐

