我在使用 SQLAlchemy 和 Pydantic 以及 Postgres 数据库的基于 FastAPI 的应用程序中遇到了一个奇怪的问题。
用户模型包含两个不同的一对多关系。由于未知的原因,第一个关系总是自动加载,尽管它不应该加载。第二个关系按预期运作。
我的模特:
class User(Base):
__tablename__ = "user"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid7, unique=True, nullable=False)
email = Column(String, unique=True, index=True)
hashed_password = Column(String)
is_active = Column(Boolean, default=True)
items = relationship("Item", back_populates="owner", lazy="select")
datastreams = relationship("DataStream", back_populates="owner", lazy="select")
class Item(Base):
__tablename__ = "item"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid7, unique=True, nullable=False)
title = Column(String, index=True)
description = Column(String, index=True)
owner_id = Column(UUID(as_uuid=True), ForeignKey("user.id"))
owner = relationship("User", back_populates="items")
class DataStream(Base):
__tablename__ = "data_stream"
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid7, unique=True, nullable=False)
name = Column(String, index=True)
description = Column(String, index=True)
type = Column(data_stream_type_enum, nullable=False) # Enum for type: 'folder' or 'stream'
parent_id = Column(UUID(as_uuid=True), ForeignKey("data_stream.id"), nullable=True) # Self-referencing foreign key
owner_id = Column(UUID(as_uuid=True), ForeignKey("user.id"))
owner = relationship("User", back_populates="datastreams")
# Self-referential relationship for parent/child hierarchy
parent = relationship("DataStream", remote_side=[id], backref="children")
我的 CRUD 层 包含以下两个功能:
def get_user(db: Session, user_id: UUID):
return db.query(models.User).filter(models.User.id == user_id).first()
def get_users(db: Session, skip: int = 0, limit: int = 100):
return db.query(models.User).offset(skip).limit(limit).all()
API 端点将使用以下架构进行响应:
from pydantic import BaseModel
class UserBase(BaseModel):
email: str
class User(UserBase):
id: UUID
is_active: bool
items: list[Item] = []
dataStreams: list[DataStream] = []
class Config:
orm_mode = True
端点看起来像这样:
@router.get("/", response_model=list[user_schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
users = crud.get_users(db, skip=skip, limit=limit)
return users
@router.get("/{user_id}", response_model=user_schemas.User)
def read_user(user_id: UUID, db: Session = Depends(get_db)):
db_user = crud.get_user(db, user_id=user_id)
if db_user is None:
raise HTTPException(status_code=404, detail="User not found")
return db_user
API 响应始终包含完全加载的项目关系,但不包含数据流关系,尽管它们存在并且与相同的所有者 ID 关联。
[
{
"email": “xxx@yyy”,
"id": "066fbcc7-61ff-7359-8000-440710dffadc",
"is_active": true,
"items": [
{
"title": "string",
"description": "string",
"id": "066fc123-237d-73e4-8000-bc6b98e36cfa",
"owner_id": "066fbcc7-61ff-7359-8000-440710dffadc"
},
{
"title": "string",
"description": "string",
"id": "066fc149-10bc-7569-8000-a629cd4d7e52",
"owner_id": "066fbcc7-61ff-7359-8000-440710dffadc"
}
],
"dataStreams": []
}
]
我的问题是:
我发现了一个奇怪的解决方案:
为了不检索一对多关系的嵌套对象,我编写了另一个不包含此类属性的架构:
class User(UserBase):
id: UUID
is_active: bool
class Config:
orm_mode = True
class UserComplex(User):
items: list[Item] = []
datastreams: List["DataStream"] = []
class Config:
orm_mode = True
通过这种方法,我将有两个不同的端点:
值得注意的是@snakecharmerb 的评论,我写了一次“dataStreams”和一次“datastreams”。 Python 并没有阻止我这样做,但它对 Pydantic 有影响。