yangy
107 lines
3.9 KiB
Python
107 lines
3.9 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.orm import Session
|
|
from typing import List, Optional
|
|
from database import get_db
|
|
from models.brand import Brand
|
|
from schemas import BrandCreate, BrandUpdate, BrandResponse
|
|
from response import ApiResponse, PageData
|
|
|
|
router = APIRouter(prefix="/api/brands", tags=["brands"])
|
|
|
|
|
|
@router.get("/", response_model=ApiResponse)
|
|
def get_brands(
|
|
skip: int = Query(0, ge=0, description="跳过记录数"),
|
|
limit: int = Query(100, ge=1, le=1000, description="返回记录数"),
|
|
name: Optional[str] = Query(None, description="品牌名称（支持模糊查询）"),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""获取所有品牌列表（支持按名称模糊查询）"""
|
|
try:
|
|
query = db.query(Brand)
|
|
if name:
|
|
query = query.filter(Brand.name.like(f"%{name}%"))
|
|
|
|
total = query.count()
|
|
brands = query.offset(skip).limit(limit).all()
|
|
|
|
if not brands:
|
|
return ApiResponse(code=2, msg="empty", data=None)
|
|
|
|
# 转换为字典列表
|
|
brands_data = [brand.to_dict() for brand in brands]
|
|
|
|
return ApiResponse(code=1, msg="success", data={"items": brands_data, "total": total, "skip": skip, "limit": limit})
|
|
except Exception as e:
|
|
return ApiResponse(code=0, msg="error", data=None)
|
|
|
|
|
|
@router.get("/{brand_id}", response_model=ApiResponse)
|
|
def get_brand(brand_id: int, db: Session = Depends(get_db)):
|
|
"""获取单个品牌"""
|
|
try:
|
|
brand = db.query(Brand).filter(Brand.id == brand_id).first()
|
|
if not brand:
|
|
return ApiResponse(code=2, msg="empty", data=None)
|
|
return ApiResponse(code=1, msg="success", data=brand.to_dict())
|
|
except Exception as e:
|
|
return ApiResponse(code=0, msg="error", data=None)
|
|
|
|
|
|
@router.post("/", response_model=ApiResponse)
|
|
def create_brand(brand: BrandCreate, db: Session = Depends(get_db)):
|
|
"""创建新品牌"""
|
|
try:
|
|
# 检查是否已存在
|
|
existing = db.query(Brand).filter(Brand.name == brand.name).first()
|
|
if existing:
|
|
return ApiResponse.error(msg="品牌名称已存在", code=0)
|
|
|
|
db_brand = Brand(name=brand.name)
|
|
db.add(db_brand)
|
|
db.commit()
|
|
db.refresh(db_brand)
|
|
return ApiResponse.success(data=db_brand.to_dict(), msg="品牌创建成功")
|
|
except Exception as e:
|
|
db.rollback()
|
|
return ApiResponse.error(msg=f"创建失败：{str(e)}", code=0)
|
|
|
|
|
|
@router.put("/{brand_id}", response_model=ApiResponse)
|
|
def update_brand(brand_id: int, brand: BrandUpdate, db: Session = Depends(get_db)):
|
|
"""更新品牌"""
|
|
try:
|
|
db_brand = db.query(Brand).filter(Brand.id == brand_id).first()
|
|
if not db_brand:
|
|
return ApiResponse.no_data(msg="品牌不存在")
|
|
|
|
# 如果更新名称，检查是否冲突
|
|
if brand.name and brand.name != db_brand.name:
|
|
existing = db.query(Brand).filter(Brand.name == brand.name).first()
|
|
if existing:
|
|
return ApiResponse.error(msg="品牌名称已存在", code=0)
|
|
db_brand.name = brand.name
|
|
|
|
db.commit()
|
|
db.refresh(db_brand)
|
|
return ApiResponse.success(data=db_brand.to_dict(), msg="品牌更新成功")
|
|
except Exception as e:
|
|
db.rollback()
|
|
return ApiResponse.error(msg=f"更新失败：{str(e)}", code=0)
|
|
|
|
|
|
@router.delete("/{brand_id}", response_model=ApiResponse)
|
|
def delete_brand(brand_id: int, db: Session = Depends(get_db)):
|
|
"""删除品牌"""
|
|
try:
|
|
db_brand = db.query(Brand).filter(Brand.id == brand_id).first()
|
|
if not db_brand:
|
|
return ApiResponse.no_data(msg="品牌不存在")
|
|
|
|
db.delete(db_brand)
|
|
db.commit()
|
|
return ApiResponse.success(msg="删除成功")
|
|
except Exception as e:
|
|
db.rollback()
|
|
return ApiResponse.error(msg=f"删除失败：{str(e)}", code=0)
|