plantillas-proyectos/backend/api/v1/modules/a76/transportation/vehicles/services.py

from typing import Optional, Tuple, List, Dict, Any

from sqlalchemy.orm import Session
from sqlalchemy import text

from . import dto, models
from api.v1.modules.a76.transportation.catalog_parity import (
    vehicle_fields_to_csv_row,
    validate_vehicle_row_for_api,
)

VEHICLE_ID_SEQ = "a76.vehicle_vehicle_id_seq"


def allocate_vehicle_id(db: Session) -> int:
    """Next surrogate vehicle_id (sequence from migration ca7d3c4e8b2a)."""
    return db.execute(text(f"SELECT nextval('{VEHICLE_ID_SEQ}')")).scalar()


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"""
        data = vehicle_data.model_dump()
        validate_vehicle_row_for_api(
            db,
            tenant_id,
            company_id,
            vehicle_fields_to_csv_row(data),
            is_update=False,
            existing_vehicle_keys=set(),
        )
        if data.get("vehicle_id") is None:
            data["vehicle_id"] = allocate_vehicle_id(db)
        new_vehicle = models.Vehicle(
            **data, 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"})
        merged = {
            "vehicle_key": vehicle.vehicle_key,
            "ace_vehicle_key": vehicle.ace_vehicle_key,
            "transporter_key": vehicle.transporter_key,
            "transport_identifier": vehicle.transport_identifier,
            "transport_type": vehicle.transport_type,
            "entity_code": vehicle.entity_code,
            "transponder_number": vehicle.transponder_number,
            "dot_number": vehicle.dot_number,
            "plate_number": vehicle.plate_number,
            "city": vehicle.city,
            "state": vehicle.state,
            "country": vehicle.country,
            "seal": vehicle.seal,
            "insurance_company_name": vehicle.insurance_company_name,
            "insurance_number": vehicle.insurance_number,
            "insurance_amount": float(vehicle.insurance_amount)
            if vehicle.insurance_amount is not None
            else None,
            "insurance_date": vehicle.insurance_date,
            "box_number": vehicle.box_number,
            "brand": vehicle.brand,
            "year": vehicle.year,
            "series": vehicle.series,
            "description": vehicle.description,
            "engine_number": vehicle.engine_number,
            "sct_permission": vehicle.sct_permission,
            "color": vehicle.color,
            "container_key": vehicle.container_key,
        }
        merged.update(update_data)
        validate_vehicle_row_for_api(
            db,
            tenant_id,
            company_id,
            vehicle_fields_to_csv_row(merged),
            is_update=True,
            existing_vehicle_keys={vehicle_key.strip()},
        )
        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