50 lines
1.6 KiB
Python
50 lines
1.6 KiB
Python
"""
|
|
仓库管理路由模块
|
|
|
|
提供仓库信息的管理功能，包括：
|
|
- 仓库列表查询（按默认仓库排序）
|
|
- 创建新仓库（自动生成仓库编码）
|
|
|
|
路由前缀: /api/warehouses
|
|
"""
|
|
from fastapi import APIRouter, Depends
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
from typing import 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, Warehouse
|
|
from .schemas import WarehouseCreate, WarehouseResponse
|
|
|
|
router = APIRouter(prefix="/warehouses", tags=["仓库管理"])
|
|
|
|
|
|
@router.get("", response_model=List[WarehouseResponse])
|
|
async def list_warehouses(
|
|
db_session: AsyncSession = Depends(get_db_session),
|
|
current_user: User = Depends(get_current_active_user)
|
|
):
|
|
result = await db_session.execute(
|
|
select(Warehouse).where(Warehouse.is_active == True).order_by(Warehouse.is_default.desc())
|
|
)
|
|
return [WarehouseResponse.from_orm(w) for w in result.scalars().all()]
|
|
|
|
|
|
@router.post("", response_model=WarehouseResponse, status_code=201)
|
|
async def create_warehouse(
|
|
warehouse_data: WarehouseCreate,
|
|
db_session: AsyncSession = Depends(get_db_session),
|
|
current_user: User = Depends(get_current_active_user)
|
|
):
|
|
data = warehouse_data.dict()
|
|
if not data.get("code"):
|
|
data["code"] = f"W{datetime.now().strftime('%Y%m%d%H%M%S')}"
|
|
|
|
warehouse = Warehouse(**data)
|
|
db_session.add(warehouse)
|
|
await db_session.commit()
|
|
await db_session.refresh(warehouse)
|
|
return WarehouseResponse.from_orm(warehouse)
|