如何通过唯一user_id从多张继承抽象模型的表中查询单条数据？

1. 修正抽象模型的多态配置 #

直接查询抽象模型报错的核心原因是SQLAlchemy无法识别该模型对应的实际数据表，需要为抽象模型配置多态映射规则：

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base, AbstractConcreteBase, declared_attr

Base = declarative_base()

# 抽象账户模型
class AbstractAccount(AbstractConcreteBase, Base):
    __abstract__ = True
    id = Column(Integer, primary_key=True, unique=True)
    name = Column(String(50))
    
    # 定义多态鉴别字段，用于区分不同子类
    @declared_attr
    def type(cls):
        return Column(String(20), nullable=False)
    
    __mapper_args__ = {
        'polymorphic_on': type,
        'polymorphic_identity': None  # 抽象模型本身无具体身份
    }

# 抽象物品模型
class AbstractItem(AbstractConcreteBase, Base):
    __abstract__ = True
    id = Column(Integer, primary_key=True, unique=True)
    item_name = Column(String(100))
    
    @declared_attr
    def type(cls):
        return Column(String(20), nullable=False)
    
    __mapper_args__ = {
        'polymorphic_on': type,
        'polymorphic_identity': None
    }

2. 子类模型配置唯一身份标识 #

每个继承自抽象模型的子类，必须指定唯一的polymorphic_identity，与抽象模型的type字段关联：

from sqlalchemy import Float, Date

# 普通用户账户
class RegularAccount(AbstractAccount):
    __tablename__ = 'regular_accounts'
    email = Column(String(100))
    
    __mapper_args__ = {
        'polymorphic_identity': 'regular'
    }

# VIP用户账户
class VipAccount(AbstractAccount):
    __tablename__ = 'vip_accounts'
    membership_level = Column(Integer)
    
    __mapper_args__ = {
        'polymorphic_identity': 'vip'
    }

# 实物物品
class PhysicalItem(AbstractItem):
    __tablename__ = 'physical_items'
    weight = Column(Float)
    
    __mapper_args__ = {
        'polymorphic_identity': 'physical'
    }

# 虚拟物品
class VirtualItem(AbstractItem):
    __tablename__ = 'virtual_items'
    expiration_date = Column(Date)
    
    __mapper_args__ = {
        'polymorphic_identity': 'virtual'
    }

3. FastAPI中实现单ID查询 #

配置完成后，可直接通过抽象模型执行查询，SQLAlchemy会自动关联所有子类表并返回对应数据：

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy.orm import Session
from database import SessionLocal  # 替换为你的数据库会话配置

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 查询账户接口
@app.get("/account/{account_id}")
def get_account(account_id: int, db: Session = Depends(get_db)):
    account = db.query(AbstractAccount).filter(AbstractAccount.id == account_id).first()
    if not account:
        raise HTTPException(status_code=404, detail="Account not found")
    return account

# 查询物品接口
@app.get("/item/{item_id}")
def get_item(item_id: int, db: Session = Depends(get_db)):
    item = db.query(AbstractItem).filter(AbstractItem.id == item_id).first()
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

关键说明 #

• AbstractConcreteBase：让SQLAlchemy允许直接查询抽象模型，自动生成包含所有子类表的UNION查询语句。
• 多态鉴别字段type：必须在抽象模型中定义，作为区分不同子类的标识，确保查询时能精准匹配对应数据表。
• 扩展性：新增子类时，只需添加表结构和唯一的polymorphic_identity，无需修改现有查询逻辑，完全满足扩展需求。

内容的提问来源于stack exchange，提问作者DJsega1