acazares
aa35635397
- Added a data table for displaying customs brokers with columns for key, name, type, license, RFC, email, phone, city, and actions. - Created a dialog for adding new customs brokers with a form for inputting necessary details. - Implemented actions for viewing details and deleting customs brokers with confirmation dialogs. - Integrated search functionality to find customs brokers by their key. - Established server-side loading for the customs brokers page to handle authentication and data retrieval.
107 lines
3.1 KiB
Python
107 lines
3.1 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("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,
|
|
company_id: int,
|
|
vehicle_data: dto.VehicleUpdateDTO,
|
|
) -> 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
|