111 lines
3.3 KiB
Python
111 lines
3.3 KiB
Python
from typing import Optional, Tuple, List, Dict, Any
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from . import dto, models
|
|
|
|
|
|
class VehicleService:
|
|
"""Service for Vehicle CRUD operations with tenant support"""
|
|
|
|
@staticmethod
|
|
def get_all(
|
|
db: Session,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
skip: int = 0,
|
|
limit: int = 50,
|
|
filters: Optional[Dict[str, Any]] = None,
|
|
) -> Tuple[List[models.Vehicle], int]:
|
|
"""Get all vehicles for a tenant/company with pagination"""
|
|
query = db.query(models.Vehicle).filter(
|
|
models.Vehicle.tenant_id == tenant_id,
|
|
models.Vehicle.company_id == company_id,
|
|
)
|
|
|
|
# Apply filters if provided
|
|
if filters:
|
|
if filters.get("vehicle_key"):
|
|
query = query.filter(
|
|
models.Vehicle.vehicle_key.ilike(f"%{filters['vehicle_key']}%")
|
|
)
|
|
if filters.get("plate_number"):
|
|
query = query.filter(
|
|
models.Vehicle.plate_number.ilike(f"%{filters['plate_number']}%")
|
|
)
|
|
if filters.get("transport_type"):
|
|
query = query.filter(
|
|
models.Vehicle.transport_type == filters["transport_type"]
|
|
)
|
|
|
|
total = query.count()
|
|
vehicles = query.offset(skip).limit(limit).all()
|
|
|
|
return vehicles, total
|
|
|
|
@staticmethod
|
|
def get_by_id(
|
|
db: Session, vehicle_key: str, tenant_id: int, company_id: int
|
|
) -> Optional[models.Vehicle]:
|
|
"""Get vehicle by vehicle_key"""
|
|
return (
|
|
db.query(models.Vehicle)
|
|
.filter(
|
|
models.Vehicle.vehicle_key == vehicle_key,
|
|
models.Vehicle.tenant_id == tenant_id,
|
|
models.Vehicle.company_id == company_id,
|
|
)
|
|
.first()
|
|
)
|
|
|
|
@staticmethod
|
|
def create(
|
|
db: Session,
|
|
vehicle_data: dto.VehicleCreateDTO,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
) -> models.Vehicle:
|
|
"""Create a new vehicle"""
|
|
new_vehicle = models.Vehicle(
|
|
**vehicle_data.model_dump(), tenant_id=tenant_id, company_id=company_id
|
|
)
|
|
db.add(new_vehicle)
|
|
db.commit()
|
|
db.refresh(new_vehicle)
|
|
return new_vehicle
|
|
|
|
@staticmethod
|
|
def update(
|
|
db: Session,
|
|
vehicle_key: str,
|
|
tenant_id: int,
|
|
vehicle_data: dto.VehicleUpdateDTO,
|
|
company_id: int,
|
|
) -> Optional[models.Vehicle]:
|
|
"""Update a vehicle"""
|
|
vehicle = VehicleService.get_by_id(db, vehicle_key, tenant_id, company_id)
|
|
if not vehicle:
|
|
return None
|
|
|
|
# Update fields (excluding vehicle_key as it's the primary key)
|
|
update_data = vehicle_data.model_dump(exclude_unset=True, exclude={"vehicle_key"})
|
|
for field, value in update_data.items():
|
|
setattr(vehicle, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(vehicle)
|
|
return vehicle
|
|
|
|
@staticmethod
|
|
def delete(
|
|
db: Session, vehicle_key: str, tenant_id: int, company_id: int
|
|
) -> bool:
|
|
"""Delete a vehicle"""
|
|
vehicle = VehicleService.get_by_id(db, vehicle_key, tenant_id, company_id)
|
|
if not vehicle:
|
|
return False
|
|
|
|
db.delete(vehicle)
|
|
db.commit()
|
|
return True
|