【问题标题】:Getting nested (joined) tables to display in the OpenAPI interface provided by FastAPI and SQLModel在 FastAPI 和 SQLModel 提供的 OpenAPI 接口中获取嵌套(连接)表
【发布时间】:2022-08-16 19:03:45
【问题描述】:

我无法理解如何使用 FastAPI 和 SQLModel 以一对多的关系显示子数据。我正在使用 Python 3.10.3、FastAPI 版本 0.78.0 和 SQLModel 版本 0.0.6。这是父/子数据库模型的简化版本:

from datetime import datetime
from email.policy import default
from sqlalchemy import UniqueConstraint
from sqlmodel import Field, SQLModel, Relationship

class CustomerBase(SQLModel):
    __table_args__ = (UniqueConstraint(\"email\"),)

    first_name: str
    last_name: str
    email: str
    active: bool | None = True

class Customer(CustomerBase, table=True):
    id: int | None =Field(primary_key=True, default=None)

class CustomerCreate(CustomerBase):
    pass

class CustomerRead(CustomerBase):
    id: int

class CustomerReadWithCalls(CustomerRead):
    calls: list[\"CallRead\"] = []

class CallBase(SQLModel):
    duration: int
    cost_per_minute: int | None = None
    customer_id: int | None = Field(default=None, foreign_key=\"customer.id\")
    created: datetime = Field(nullable=False, default=datetime.now().date())

class Call(CallBase, table=True):
    id: int | None = Field(primary_key=True)

class CallCreate(CallBase):
    pass

class CallRead(CallBase):
    id: int

class CallReadWithCustomer(CallRead):
    customer: CustomerRead | None

这是API路线:

from fastapi import APIRouter, HTTPException, Depends, Query
from rbi_app.crud.customer import (
    get_customers,
    get_customer,
)
from rbi_app.models import (
    CustomerRead,
    CustomerReadWithCalls,
)
from rbi_app.database import Session, get_session

router = APIRouter()

@router.get(\"/customers/\", status_code=200, response_model=list[CustomerRead])
def read_customers(
    email: str = \"\",
    offset: int = 0,
    limit: int = Query(default=100, lte=100),
    db: Session = Depends(get_session)
):
    return get_customers(db, email, offset=offset, limit=limit)

@router.get(\"/customers/{customer_id}\", status_code=200, response_model=CustomerReadWithCalls)
def read_customer(id: int, db: Session = Depends(get_session)):
    customer = get_customer(db, id)
    if customer is None:
        raise HTTPException(status_code=404, detail=f\"Customer not found for {id=}\")
    return customer

以下是 API Route 端点对数据库的查询:

from sqlmodel import select
from rbi_app.database import Session
from rbi_app.models import (
    Customer,
    CustomerCreate,
)
# from rbi_app.schemas.customer import CustomerCreate
    
def get_customer(db: Session, id: int):
    return db.get(Customer, id)
    
def get_customers(db: Session, email: str = \"\", offset: int = 0, limit: int = 100):
    if email:
        return db.exec(select(Customer).where(Customer.email == email)).first()
    return db.exec(select(Customer).offset(offset).limit(limit).order_by(Customer.id)).all()

当我导航到获取所有客户的路线时,我的查询运行并获得了客户,但客户中没有“呼叫”列表属性。 OpenAPI 显示显示“调用”属性,但它是空的。

我究竟做错了什么?任何建议或参考将不胜感激!

  • 如果没有 API 路由部分,就很难说出哪里出了问题。此外,我们缺少查询,但我猜 pydantic 使用空数组初始化 calls 字段,因为您没有将 calls 提取/添加到响应中。
  • 也许是因为调用不是急切加载的。查看 SQLModel 问题中的this comment 以获取更多信息。
  • 你是对的,我应该包括 API 路由和实际查询。更改了 OP 以更改此设置。

标签: python fastapi pydantic sqlmodel


【解决方案1】:

这里的问题似乎是您没有在Customer 模型(或Call 模块)上定义关系。由于您使用Customer 模型查询数据库并且它没有calls 属性,因此get_customer 函数返回的对象中不存在这些数据。

即使您的路由将 CustomerReadWithCalls 定义为响应模型,在调用它时,该类的对象也只能使用您的路由处理函数返回的数据进行实例化,在这种情况下是您的 Customer 实例。因为它甚至没有calls 属性(更不用说数据),所以CustomerReadWithCalls 对象本质上是使用您为calls 字段定义的默认值创建的——空列表。

添加

    calls: list["Call"] = Relationship(back_populates="customer")

给你的Customer 模型应该足够。

(但作为旁注,对我来说,当我在 CallRead 定义之后明确更新 CustomerReadWithCalls 模型上的引用时,路线文档才能正常工作。)

这是一个完整的工作示例。

models.py

from datetime import datetime

from sqlalchemy import UniqueConstraint
from sqlmodel import Field, Relationship, SQLModel


class CustomerBase(SQLModel):
    __table_args__ = (UniqueConstraint("email"),)

    first_name: str
    last_name: str
    email: str
    active: bool | None = True


class Customer(CustomerBase, table=True):
    id: int | None = Field(primary_key=True, default=None)

    calls: list["Call"] = Relationship(back_populates="customer")


class CustomerCreate(CustomerBase):
    pass


class CustomerRead(CustomerBase):
    id: int


class CustomerReadWithCalls(CustomerRead):
    calls: list["CallRead"] = []


class CallBase(SQLModel):
    duration: int
    cost_per_minute: int | None = None
    customer_id: int | None = Field(default=None, foreign_key="customer.id")
    created: datetime = Field(nullable=False, default=datetime.now().date())


class Call(CallBase, table=True):
    id: int | None = Field(primary_key=True, default=None)

    customer: Customer | None = Relationship(back_populates="calls")


class CallCreate(CallBase):
    pass


class CallRead(CallBase):
    id: int


# After the definition of `CallRead`, update the forward reference to it:
CustomerReadWithCalls.update_forward_refs()


class CallReadWithCustomer(CallRead):
    customer: CustomerRead | None

routes.py

from fastapi import FastAPI, HTTPException, Depends
from sqlmodel import Session, SQLModel, create_engine

from .models import CustomerReadWithCalls, Customer, Call


api = FastAPI()

sqlite_file_name = 'database.db'
sqlite_url = f'sqlite:///{sqlite_file_name}'
engine = create_engine(sqlite_url, echo=True)


@api.on_event('startup')
def initialize_db():
    SQLModel.metadata.drop_all(engine)
    SQLModel.metadata.create_all(engine)

    # For testing:
    with Session(engine) as session:
        customer = Customer(first_name="Foo", last_name="Bar", email="foo@bar.com")
        call1 = Call(duration=123)
        call2 = Call(duration=456)
        customer.calls.extend([call1, call2])
        session.add(customer)
        session.commit()


def get_session() -> Session:
    session = Session(engine)
    try:
        yield session
    finally:
        session.close()


def get_customer(db: Session, id: int):
    return db.get(Customer, id)


@api.get("/customers/{customer_id}", status_code=200, response_model=CustomerReadWithCalls)
def read_customer(customer_id: int, db: Session = Depends(get_session)):
    customer = get_customer(db, customer_id)
    if customer is None:
        raise HTTPException(status_code=404, detail=f"Customer not found for {customer_id=}")
    return customer

启动 API 服务器并将GET 发送到http://127.0.0.1:8000/customers/1 给了我

{
  "first_name": "Foo",
  "last_name": "Bar",
  "email": "foo@bar.com",
  "active": true,
  "id": 1,
  "calls": [
    {
      "duration": 123,
      "cost_per_minute": null,
      "customer_id": 1,
      "created": "2022-08-16T00:00:00",
      "id": 1
    },
    {
      "duration": 456,
      "cost_per_minute": null,
      "customer_id": 1,
      "created": "2022-08-16T00:00:00",
      "id": 2
    }
  ]
}

希望这可以帮助。

【讨论】:

    猜你喜欢
    • 2015-02-16
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2014-10-18
    • 2016-07-14
    • 1970-01-01
    • 1970-01-01
    • 2016-11-25
    相关资源
    最近更新 更多