SQLAlchemy如何实现公寓关联当前入住居民的一对一关系查询
完全可以通过SQLAlchemy的原生能力实现,你可以通过带过滤条件的自定义关系+属性映射来实现需求,具体步骤如下:
1. 调整Apartment数据模型
新增带业务过滤条件的一对一关系,直接映射当前入住住户:
from sqlalchemy import and_, func from sqlalchemy.ext.hybrid import hybrid_property class Apartment(Base): id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, unique=True, nullable=False) address = Column(String(200)) residents = relationship("Occupation") # 新增当前入住记录的一对一关联 current_occupation = relationship( "Occupation", primaryjoin=and_( id == Occupation.apartment_id, # 当前入住的判断规则可根据业务调整,示例为「结束日期为空且开始日期早等于当前日期」 Occupation.end == None, Occupation.start <= func.current_date() ), uselist=False, # 指定为一对一关系,不返回列表 viewonly=True # 标记为只读,避免误修改关联数据 ) # 新增属性直接返回当前住户,省略一层occupation嵌套 @hybrid_property def current_resident(self): return self.current_occupation.resident if self.current_occupation else None
可选:增加数据一致性约束
为了避免同一公寓出现多条有效入住记录,可在Occupation模型新增部分唯一索引:
class Occupation(Base): # 原有字段保持不变 __table_args__ = ( # 未结束的入住记录,同一公寓仅允许存在一条 UniqueConstraint('apartment_id', 'end', name='unique_active_occupation_per_apartment'), )
2. 修改查询语句
调整关联加载逻辑,加载新增的当前入住关联即可:
db.query(models.Apartment).options( joinedload(models.Apartment.current_occupation).joinedload(models.Occupation.resident), ).all()
3. 调整Pydantic Schema
直接映射新增的current_resident属性即可输出你需要的结构:
from pydantic import BaseModel from uuid import UUID from typing import Optional class ResidentSchema(BaseModel): id: UUID name: str class Config: orm_mode = True class ApartmentSchema(BaseModel): id: UUID address: str current_resident: Optional[ResidentSchema] class Config: orm_mode = True
内容的提问来源于stack exchange,提问作者Wouter Goossens
相关产品推荐
相关产品推荐

