init

src/api/inventory/supplier_routes.py

@@ -0,0 +1,54 @@
+"""
+供应商管理路由模块
+
+提供供应商信息的管理功能，包括：
+- 供应商列表查询（支持分页、搜索）
+- 创建新供应商（自动生成供应商编码）
+
+路由前缀: /api/suppliers
+"""
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select
+from typing import Optional, List
+from datetime import datetime
+
+from database.database import get_db_session
+from services.auth_service import get_current_active_user
+from models.database import User, Supplier
+from .schemas import SupplierCreate, SupplierResponse
+
+router = APIRouter(prefix="/suppliers", tags=["供应商管理"])
+
+
+@router.get("", response_model=List[SupplierResponse])
+async def list_suppliers(
+    skip: int = Query(0, ge=0),
+    limit: int = Query(20, ge=1, le=100),
+    search: Optional[str] = None,
+    db_session: AsyncSession = Depends(get_db_session),
+    current_user: User = Depends(get_current_active_user)
+):
+    query = select(Supplier).where(Supplier.is_active == True)
+    if search:
+        query = query.where(Supplier.name.ilike(f"%{search}%"))
+    query = query.offset(skip).limit(limit).order_by(Supplier.created_at.desc())
+    result = await db_session.execute(query)
+    return [SupplierResponse.from_orm(s) for s in result.scalars().all()]
+
+
+@router.post("", response_model=SupplierResponse, status_code=201)
+async def create_supplier(
+    supplier_data: SupplierCreate,
+    db_session: AsyncSession = Depends(get_db_session),
+    current_user: User = Depends(get_current_active_user)
+):
+    data = supplier_data.dict()
+    if not data.get("code"):
+        data["code"] = f"S{datetime.now().strftime('%Y%m%d%H%M%S')}"
+    
+    supplier = Supplier(**data)
+    db_session.add(supplier)
+    await db_session.commit()
+    await db_session.refresh(supplier)
+    return SupplierResponse.from_orm(supplier)