03. 使用FastAPI实现查询API接口

def get_db():
    """提供数据库会话：成功则 commit，异常则 rollback"""
    db = SessionLocal()
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

class GoodsSchema(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int | None = None
    name: str | None = None
    price: Decimal | None = None
    stock: int | None = None
    category_id: int | None = None
    category_name: str | None = None

from fastapi import APIRouter, Depends
from sqlalchemy import select, or_
from sqlalchemy.orm import Session

from app.db.session import get_db
from app.models.category import Category
from app.models.goods import Goods
from app.schemas.goods import GoodsSchema
from app.schemas.response import Result

# /api/goods/all
router = APIRouter(prefix="/goods")

@router.get("/all")
def get_all_goods(keyword: str | None = None, db: Session = Depends(get_db)):
    """查询所有的商品列表"""
    result = []
    stmt = (
        select(Goods, Category.name)
        .outerjoin(Category, Category.id == Goods.category_id)
    )
    # select * from goods left join category on goods.category_id = category.id where goods.name like %豆%
    # order by goods.id desc
    if keyword:
        pattern = f"%{keyword}%"
        stmt = stmt.where(  # 要重新赋值
            or_(Goods.name.ilike(pattern), Category.name.ilike(pattern))
        )
    stmt = stmt.order_by(Goods.id.desc())
    rows = db.execute(stmt).all()
    for goods, category_name in rows:
        goods_schema = GoodsSchema.model_validate(goods)
        goods_schema.category_name = category_name
        result.append(goods_schema)
    return Result.success(data=result)

from typing import Any, List

from pydantic import BaseModel

class Result(BaseModel):
    """通用返回对象"""

    code: int = 0
    msg: str = "请求成功"
    data: Any = None

    @staticmethod
    def success(data=None, msg: str = "操作成功") -> "Result":
        return Result(code=0, msg=msg, data=data)

    @staticmethod
    def error(code: int = 500, msg: str = "操作失败") -> "Result":
        return Result(code=code, msg=msg)

class PageResult(BaseModel):
    """分页结果"""
    list: List[Any]
    total: int