04. 使用FastAPI实现增删改查API接口

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

from app.db.session import get_db
from app.schemas.goods import GoodsSchema
from app.schemas.response import Result
from app.services import goods_service

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

# /api/goods/create
@router.post("/create")
def create_goods(goods: GoodsSchema, db: Session = Depends(get_db)):
    """新增商品"""
    result = goods_service.create_goods(goods, db)
    return Result.success(data=result)

@router.put("/update/{goods_id}")
def update_goods(goods_id: int, goods: GoodsSchema, db: Session = Depends(get_db)):
    """更新商品"""
    result = goods_service.update_goods(goods_id, goods, db)
    return Result.success(data=result)

@router.delete("/delete/{goods_id}")
def delete_goods(goods_id: int, db: Session = Depends(get_db)):
    """删除商品"""
    goods_service.delete_goods(goods_id, db)
    return Result.success()

#  /api/goods/all
@router.get("/all")
def get_all_goods(keyword: str | None = None, db: Session = Depends(get_db)):
    """查询所有的商品列表"""
    result = goods_service.get_all_goods(keyword, db)
    return Result.success(data=result)

@router.get("/page")
def get_page_goods(keyword: str | None = None, page_num: int = 1, page_size: int = 10, db: Session = Depends(get_db)):
    """分页查询商品列表"""
    page_result = goods_service.get_page_goods(keyword, page_num, page_size, db)
    return Result.success(data=page_result)

from sqlalchemy import select, or_, func
from sqlalchemy.orm import Session

from app.core.exceptions import BizException
from app.models.category import Category
from app.models.goods import Goods
from app.schemas.goods import GoodsSchema
from app.schemas.response import PageResult

def create_goods(goods: GoodsSchema, db: Session):
    """新增商品"""
    if goods.name is None:
        raise BizException(code=500, msg="名称必填")
    # 把 schema转换成字典
    payload = goods.model_dump(exclude={"id"}, exclude_none=True)
    # {name: "", price: 0.99}
    db_goods = Goods(**payload)  # name=xxx, price=xxx
    db.add(db_goods)
    db.flush()
    return GoodsSchema.model_validate(db_goods).model_dump()

def update_goods(goods_id: int, goods: GoodsSchema, db: Session):
    """更新商品"""
    db_goods = db.scalar(select(Goods).where(Goods.id == goods_id))
    if db_goods is None:
        raise BizException(code=500, msg="商品不存在")
    payload = goods.model_dump(exclude={"id"}, exclude_unset=True)
    for key, value in payload.items():
        setattr(db_goods, key, value)
    db.flush()
    return GoodsSchema.model_validate(db_goods).model_dump()

def delete_goods(goods_id: int, db: Session):
    """删除商品"""
    db_goods = db.scalar(select(Goods).where(Goods.id == goods_id))
    if db_goods is None:
        raise BizException(code=500, msg="商品不存在")
    db.delete(db_goods)
    db.flush()

def get_all_goods(keyword: str, db: Session):
    """查询所有商品"""
    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

def get_page_goods(keyword: str, page_num: int, page_size: int, db: Session):
    """分页查询商品列表"""
    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
    conditions = []
    if keyword:
        pattern = f"%{keyword}%"
        conditions.append(or_(Goods.name.ilike(pattern), Category.name.ilike(pattern)))
        stmt = stmt.where(  # 要重新赋值
            *conditions  # or_(Goods.name.ilike(pattern), Category.name.ilike(pattern))
        )
    # 查询商品的总数
    total = db.scalar(select(func.count(Goods.id)).where(*conditions)) or 0  # select count(id)
    stmt = stmt.order_by(Goods.id.desc())
    stmt = stmt.offset((page_num - 1) * page_size).limit(page_size)  # 分页
    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 PageResult(total=total, list=result)

from fastapi import HTTPException, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

from app.schemas.response import Result

class BizException(Exception):
    """业务异常：抛出后由全局处理器转成 Result.error"""
    def __init__(self, code: int, msg: str):
        self.code = code
        self.msg = msg

async def biz_exception_handler(_request: Request, exc: BizException) -> JSONResponse:
    # _request 是 FastAPI 约定参数，这里用不到
    return JSONResponse(
        status_code=200,
        content=Result.error(code=exc.code, msg=exc.msg).model_dump(),
    )

async def http_exception_handler(_request: Request, exc: HTTPException) -> JSONResponse:
    """处理 HTTP 异常"""
    return JSONResponse(
        status_code=exc.status_code,
        content=Result.error(code=exc.status_code, msg=str(exc.detail)).model_dump(),
    )

async def validation_exception_handler(_request: Request, exc: RequestValidationError) -> JSONResponse:
    """处理参数校验异常"""
    errors = exc.errors()
    msg = errors[0].get("msg", "参数校验失败") if errors else "参数校验失败"
    return JSONResponse(
        status_code=422,
        content=Result.error(code=422, msg=msg).model_dump(),
    )