147 lines
4.6 KiB
Python
147 lines
4.6 KiB
Python
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
|
from sqlalchemy import func, or_
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.database import get_db
|
|
from app.core.db_auth import DbUserContext, get_db_current_user
|
|
from app.core.tenant_context import get_tenant_no
|
|
from app.core.utils import generate_uuid
|
|
from app.models.db_models import DbVehicle
|
|
from app.schemas.db.master_data import VehicleCreate, VehicleRead, VehicleUpdate
|
|
from app.services.db.mappers import vehicle_to_read
|
|
|
|
router = APIRouter(prefix="/master-data/vehicles", tags=["vehicles"])
|
|
|
|
|
|
def _base_query(db: Session, tenant_no: str):
|
|
return db.query(DbVehicle).filter(
|
|
DbVehicle.tenant_no == tenant_no,
|
|
DbVehicle.del_flag == 0,
|
|
)
|
|
|
|
|
|
@router.get("/", response_model=list[VehicleRead])
|
|
def list_vehicles(
|
|
q: Optional[str] = Query(default=None, description="模糊搜索车牌号"),
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> list[VehicleRead]:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
query = _base_query(db, tenant_no).order_by(DbVehicle.create_time.desc())
|
|
if q:
|
|
keyword = q.lower()
|
|
query = query.filter(
|
|
or_(
|
|
func.lower(DbVehicle.vehicle_no).contains(keyword),
|
|
func.lower(DbVehicle.remark).contains(keyword),
|
|
)
|
|
)
|
|
rows = query.all()
|
|
return [vehicle_to_read(row) for row in rows]
|
|
|
|
|
|
@router.get("/{vehicle_id}", response_model=VehicleRead)
|
|
def get_vehicle(
|
|
vehicle_id: str,
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> VehicleRead:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
row = _get_vehicle(db, tenant_no, vehicle_id)
|
|
return vehicle_to_read(row)
|
|
|
|
|
|
@router.post("/", response_model=VehicleRead, status_code=status.HTTP_201_CREATED)
|
|
def create_vehicle(
|
|
payload: VehicleCreate,
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> VehicleRead:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
_ensure_unique_vehicle_no(db, tenant_no, payload.vehicle_no)
|
|
row = DbVehicle(
|
|
id=generate_uuid(),
|
|
tenant_no=tenant_no,
|
|
vehicle_no=payload.vehicle_no.strip(),
|
|
remark=payload.remarks,
|
|
create_by=user.id,
|
|
create_dept_id=user.dept_id,
|
|
)
|
|
db.add(row)
|
|
db.commit()
|
|
db.refresh(row)
|
|
return vehicle_to_read(row)
|
|
|
|
|
|
@router.put("/{vehicle_id}", response_model=VehicleRead)
|
|
def update_vehicle(
|
|
vehicle_id: str,
|
|
payload: VehicleUpdate,
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> VehicleRead:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
row = _get_vehicle(db, tenant_no, vehicle_id)
|
|
if payload.vehicle_no and payload.vehicle_no.strip().lower() != row.vehicle_no.lower():
|
|
_ensure_unique_vehicle_no(db, tenant_no, payload.vehicle_no)
|
|
row.vehicle_no = payload.vehicle_no.strip()
|
|
if payload.remarks is not None:
|
|
row.remark = payload.remarks
|
|
row.update_by = user.id
|
|
db.commit()
|
|
db.refresh(row)
|
|
return vehicle_to_read(row)
|
|
|
|
|
|
@router.patch("/{vehicle_id}", response_model=VehicleRead)
|
|
def patch_vehicle(
|
|
vehicle_id: str,
|
|
payload: VehicleUpdate,
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> VehicleRead:
|
|
return update_vehicle(vehicle_id, payload, db, user)
|
|
|
|
|
|
@router.delete("/{vehicle_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_vehicle(
|
|
vehicle_id: str,
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> None:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
row = _get_vehicle(db, tenant_no, vehicle_id)
|
|
row.del_flag = 1
|
|
row.update_by = user.id
|
|
db.commit()
|
|
|
|
|
|
def _get_vehicle(db: Session, tenant_no: str, vehicle_id: str) -> DbVehicle:
|
|
row = (
|
|
_base_query(db, tenant_no)
|
|
.filter(DbVehicle.id == vehicle_id)
|
|
.one_or_none()
|
|
)
|
|
if row is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Vehicle {vehicle_id} not found",
|
|
)
|
|
return row
|
|
|
|
|
|
def _ensure_unique_vehicle_no(db: Session, tenant_no: str, vehicle_no: str) -> None:
|
|
normalized = vehicle_no.strip().lower()
|
|
exists = (
|
|
_base_query(db, tenant_no)
|
|
.filter(func.lower(DbVehicle.vehicle_no) == normalized)
|
|
.first()
|
|
)
|
|
if exists:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Vehicle number already exists",
|
|
)
|