155 lines
4.6 KiB
Python
155 lines
4.6 KiB
Python
from fastapi import APIRouter, HTTPException, Body
|
|
from typing import Optional
|
|
from decimal import Decimal
|
|
from datetime import date as date_module
|
|
|
|
from database import get_session
|
|
from models import Cost, Vehicle
|
|
from schemas import (
|
|
CostCreate,
|
|
CostUpdate,
|
|
CostDelete,
|
|
CostResponse,
|
|
COST_TYPES,
|
|
)
|
|
|
|
|
|
router = APIRouter(prefix="/costs", tags=["costs"])
|
|
|
|
|
|
def _to_cost_response(cost: Cost) -> CostResponse:
|
|
"""将 Cost 模型转换为 CostResponse"""
|
|
return CostResponse(
|
|
id=cost.id,
|
|
vehicle_id=cost.vehicle_id,
|
|
date=cost.date,
|
|
type=cost.type,
|
|
amount=float(cost.amount),
|
|
mileage=cost.mileage,
|
|
notes=cost.notes,
|
|
is_installment=cost.is_installment,
|
|
installment_months=cost.installment_months,
|
|
is_deleted=cost.is_deleted,
|
|
created_at=cost.created_at,
|
|
updated_at=cost.updated_at,
|
|
)
|
|
|
|
|
|
@router.get("/list", response_model=list[CostResponse])
|
|
def get_costs(
|
|
vehicle_id: Optional[int] = None, cost_type: Optional[str] = None, limit: int = 50
|
|
):
|
|
"""获取费用记录列表（排除已删除）"""
|
|
with get_session() as session:
|
|
query = session.query(Cost).filter(Cost.is_deleted == False)
|
|
if vehicle_id:
|
|
query = query.filter(Cost.vehicle_id == vehicle_id)
|
|
if cost_type:
|
|
query = query.filter(Cost.type == cost_type)
|
|
|
|
costs = query.order_by(Cost.date.desc()).limit(limit).all()
|
|
return [_to_cost_response(c) for c in costs]
|
|
|
|
|
|
@router.post("/create", response_model=CostResponse)
|
|
def create_cost(cost: CostCreate):
|
|
"""添加费用记录"""
|
|
with get_session() as session:
|
|
# 验证车辆存在
|
|
vehicle = session.query(Vehicle).filter(Vehicle.id == cost.vehicle_id).first()
|
|
if not vehicle:
|
|
raise HTTPException(status_code=404, detail="Vehicle not found")
|
|
|
|
# 验证费用类型
|
|
if cost.type not in COST_TYPES:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Invalid cost type. Must be one of: {', '.join(COST_TYPES)}",
|
|
)
|
|
|
|
db_cost = Cost(
|
|
vehicle_id=cost.vehicle_id,
|
|
date=cost.date,
|
|
type=cost.type,
|
|
amount=Decimal(str(cost.amount)),
|
|
mileage=cost.mileage,
|
|
notes=cost.notes or "",
|
|
is_installment=cost.is_installment or False,
|
|
installment_months=cost.installment_months or 12,
|
|
)
|
|
session.add(db_cost)
|
|
session.commit()
|
|
session.refresh(db_cost)
|
|
|
|
return _to_cost_response(db_cost)
|
|
|
|
|
|
@router.post("/update")
|
|
def update_cost(
|
|
id: int = Body(...),
|
|
type: Optional[str] = Body(None),
|
|
date: Optional[str] = Body(None),
|
|
amount: Optional[float] = Body(None),
|
|
mileage: Optional[int] = Body(None),
|
|
notes: Optional[str] = Body(None),
|
|
is_installment: Optional[bool] = Body(None),
|
|
installment_months: Optional[int] = Body(None),
|
|
):
|
|
"""更新费用记录"""
|
|
with get_session() as session:
|
|
cost = session.query(Cost).filter(Cost.id == id).first()
|
|
if not cost:
|
|
raise HTTPException(status_code=404, detail="Cost record not found")
|
|
|
|
# 验证费用类型
|
|
if type is not None:
|
|
if type not in COST_TYPES:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Invalid cost type. Must be one of: {', '.join(COST_TYPES)}",
|
|
)
|
|
cost.type = type
|
|
|
|
# 处理日期转换
|
|
if date is not None:
|
|
cost.date = date_module.fromisoformat(date)
|
|
|
|
if amount is not None:
|
|
cost.amount = Decimal(str(amount))
|
|
if mileage is not None:
|
|
cost.mileage = mileage
|
|
if notes is not None:
|
|
cost.notes = notes or ""
|
|
if is_installment is not None:
|
|
cost.is_installment = is_installment
|
|
if installment_months is not None:
|
|
cost.installment_months = installment_months
|
|
|
|
session.commit()
|
|
session.refresh(cost)
|
|
|
|
return _to_cost_response(cost)
|
|
|
|
|
|
@router.post("/delete")
|
|
def delete_cost(cost: CostDelete):
|
|
"""软删除费用记录"""
|
|
with get_session() as session:
|
|
db_cost = (
|
|
session.query(Cost)
|
|
.filter(Cost.id == cost.id, Cost.is_deleted == False)
|
|
.first()
|
|
)
|
|
if not db_cost:
|
|
raise HTTPException(status_code=404, detail="Cost record not found")
|
|
|
|
db_cost.is_deleted = True
|
|
session.commit()
|
|
return {"message": "Cost record deleted successfully"}
|
|
|
|
|
|
@router.get("/types")
|
|
def get_cost_types():
|
|
"""获取费用类型列表"""
|
|
return {"types": COST_TYPES}
|