154 lines
4.8 KiB
Python
154 lines
4.8 KiB
Python
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 (
|
|
trailer_fields_to_csv_row,
|
|
validate_trailer_row_for_api,
|
|
)
|
|
|
|
TRAILER_ID_SEQ = "a76.trailer_trailer_id_seq"
|
|
|
|
|
|
def allocate_trailer_id(db: Session) -> int:
|
|
"""Next surrogate trailer_id (sequence from migration ca7d3c4e8b2a)."""
|
|
return db.execute(text(f"SELECT nextval('{TRAILER_ID_SEQ}')")).scalar()
|
|
|
|
|
|
class TrailerService:
|
|
"""Service for Trailer 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.Trailer], int]:
|
|
"""Get all trailers for a tenant/company with pagination"""
|
|
query = db.query(models.Trailer).filter(
|
|
models.Trailer.tenant_id == tenant_id,
|
|
models.Trailer.company_id == company_id,
|
|
)
|
|
|
|
# Apply filters if provided
|
|
if filters:
|
|
if filters.get("trailer_number"):
|
|
query = query.filter(
|
|
models.Trailer.trailer_number.ilike(f"%{filters['trailer_number']}%")
|
|
)
|
|
if filters.get("plate_number"):
|
|
query = query.filter(
|
|
models.Trailer.plate_number.ilike(f"%{filters['plate_number']}%")
|
|
)
|
|
if filters.get("trailer_type_key"):
|
|
query = query.filter(
|
|
models.Trailer.trailer_type_key == filters["trailer_type_key"]
|
|
)
|
|
|
|
total = query.count()
|
|
trailers = query.offset(skip).limit(limit).all()
|
|
|
|
return trailers, total
|
|
|
|
@staticmethod
|
|
def get_by_id(
|
|
db: Session, trailer_number: str, tenant_id: int, company_id: int
|
|
) -> Optional[models.Trailer]:
|
|
"""Get trailer by trailer_number"""
|
|
return (
|
|
db.query(models.Trailer)
|
|
.filter(
|
|
models.Trailer.trailer_number == trailer_number,
|
|
models.Trailer.tenant_id == tenant_id,
|
|
models.Trailer.company_id == company_id,
|
|
)
|
|
.first()
|
|
)
|
|
|
|
@staticmethod
|
|
def create(
|
|
db: Session,
|
|
trailer_data: dto.TrailerCreateDTO,
|
|
tenant_id: int,
|
|
company_id: int,
|
|
) -> models.Trailer:
|
|
"""Create a new trailer"""
|
|
data = trailer_data.model_dump()
|
|
validate_trailer_row_for_api(
|
|
tenant_id,
|
|
company_id,
|
|
trailer_fields_to_csv_row(data),
|
|
is_update=False,
|
|
existing_trailer_numbers=set(),
|
|
)
|
|
if data.get("trailer_id") is None:
|
|
data["trailer_id"] = allocate_trailer_id(db)
|
|
new_trailer = models.Trailer(
|
|
**data, tenant_id=tenant_id, company_id=company_id
|
|
)
|
|
db.add(new_trailer)
|
|
db.commit()
|
|
db.refresh(new_trailer)
|
|
return new_trailer
|
|
|
|
@staticmethod
|
|
def update(
|
|
db: Session,
|
|
trailer_number: str,
|
|
tenant_id: int,
|
|
trailer_data: dto.TrailerUpdateDTO,
|
|
company_id: int,
|
|
) -> Optional[models.Trailer]:
|
|
"""Update a trailer"""
|
|
trailer = TrailerService.get_by_id(db, trailer_number, tenant_id, company_id)
|
|
if not trailer:
|
|
return None
|
|
|
|
# Update fields (excluding trailer_number as it's the primary key)
|
|
update_data = trailer_data.model_dump(
|
|
exclude_unset=True, exclude={"trailer_number"}
|
|
)
|
|
merged = {
|
|
"trailer_number": trailer.trailer_number,
|
|
"ace_trailer_number": trailer.ace_trailer_number,
|
|
"trailer_type_key": trailer.trailer_type_key,
|
|
"seal": trailer.seal,
|
|
"entity_code": trailer.entity_code,
|
|
"plate_number": trailer.plate_number,
|
|
"state": trailer.state,
|
|
"country": trailer.country,
|
|
"container_key": trailer.container_key,
|
|
}
|
|
merged.update(update_data)
|
|
validate_trailer_row_for_api(
|
|
tenant_id,
|
|
company_id,
|
|
trailer_fields_to_csv_row(merged),
|
|
is_update=True,
|
|
existing_trailer_numbers={trailer_number.strip()},
|
|
)
|
|
for field, value in update_data.items():
|
|
setattr(trailer, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(trailer)
|
|
return trailer
|
|
|
|
@staticmethod
|
|
def delete(
|
|
db: Session, trailer_number: str, tenant_id: int, company_id: int
|
|
) -> bool:
|
|
"""Delete a trailer"""
|
|
trailer = TrailerService.get_by_id(db, trailer_number, tenant_id, company_id)
|
|
if not trailer:
|
|
return False
|
|
|
|
db.delete(trailer)
|
|
db.commit()
|
|
return True
|