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

from sqlalchemy.orm import Session

from . import dto, models


class CustomsBrokerService:
    @staticmethod
    def get_by_id(db: Session, broker_key: str, tenant_id: int, company_id: int):
        """Get a customs broker by broker_key with tenant/company validation"""
        return (
            db.query(models.CustomsBroker)
            .filter(
                models.CustomsBroker.broker_key == broker_key,
                models.CustomsBroker.tenant_id == tenant_id,
                models.CustomsBroker.company_id == company_id,
            )
            .first()
        )

    @staticmethod
    def get_all(
        db: Session,
        tenant_id: int,
        company_id: int,
        skip: int = 0,
        limit: int = 100,
        filters: dict = None,
    ):
        """Get all customs brokers for a tenant/company with pagination"""
        query = db.query(models.CustomsBroker).filter(
            models.CustomsBroker.tenant_id == tenant_id,
            models.CustomsBroker.company_id == company_id,
        )

        total = query.count()
        items = query.offset(skip).limit(limit).all()
        return items, total

    @staticmethod
    def create(db: Session, broker_data: dto.CustomsBrokerCreateDTO, tenant_id: int, company_id: int):
        """Create a new customs broker"""
        broker_dict = broker_data.model_dump()
        broker_dict["tenant_id"] = tenant_id
        broker_dict["company_id"] = company_id
        
        new_broker = models.CustomsBroker(**broker_dict)
        db.add(new_broker)
        db.commit()
        db.refresh(new_broker)
        return new_broker

    @staticmethod
    def update(db: Session, broker_key: str, tenant_id: int, broker_data: dto.CustomsBrokerUpdateDTO, company_id: int):
        """Update an existing customs broker"""
        broker = CustomsBrokerService.get_by_id(db, broker_key, tenant_id, company_id)
        if broker:
            for key, value in broker_data.model_dump(exclude_unset=True).items():
                setattr(broker, key, value)
            db.commit()
            db.refresh(broker)
        return broker

    @staticmethod
    def delete(db: Session, broker_key: str, tenant_id: int, company_id: int):
        """Delete a customs broker"""
        broker = CustomsBrokerService.get_by_id(db, broker_key, tenant_id, company_id)
        if broker:
            db.delete(broker)
            db.commit()
            return True
        return False


class CustomsBrokerVUService:
    @staticmethod
    def get_by_broker_key(db: Session, broker_key: str):
        return (
            db.query(models.CustomsBrokerVU)
            .join(models.CustomsBroker)
            .filter(models.CustomsBroker.broker_key == broker_key)
            .first()
        )

    @staticmethod
    def create_vu(db: Session, vu_data: dto.CustomsBrokerVUCreateDTO):
        new_vu = models.CustomsBrokerVU(**vu_data.dict())
        db.add(new_vu)
        db.commit()
        db.refresh(new_vu)
        return new_vu

    @staticmethod
    def update_vu(db: Session, broker_key: str, vu_data: dto.CustomsBrokerVUCreateDTO):
        # We need the custom broker ID to insert a new VU
        broker = db.query(models.CustomsBroker).filter(models.CustomsBroker.broker_key == broker_key).first()
        if not broker:
            return None 

        vu = db.query(models.CustomsBrokerVU).filter(models.CustomsBrokerVU.customs_broker_id == broker.id).first()
        
        if vu:
            # Update existing
            for key, value in vu_data.model_dump(exclude_unset=True).items():
                setattr(vu, key, value)
            db.commit()
            db.refresh(vu)
            return vu
        else:
            # Create new
            new_vu_data = vu_data.model_dump()
            new_vu = models.CustomsBrokerVU(customs_broker_id=broker.id, **new_vu_data)
            db.add(new_vu)
            db.commit()
            db.refresh(new_vu)
            return new_vu

    @staticmethod
    def delete_vu(db: Session, broker_key: str):
        vu = CustomsBrokerVUService.get_by_broker_key(db, broker_key)
        if vu:
            db.delete(vu)
            db.commit()
        return vu


class CustomsBrokerPersonnelService:
    @staticmethod
    def get_by_broker_key_and_line(db: Session, broker_key: str, line: int):
        return (
            db.query(models.CustomsBrokerPersonnel)
            .join(models.CustomsBroker)
            .filter(
                models.CustomsBroker.broker_key == broker_key,
                models.CustomsBrokerPersonnel.line == line,
            )
            .first()
        )

    @staticmethod
    def create_personnel(db: Session, broker_key: str, personnel_data: dto.CustomsBrokerPersonnelDTO):
        broker = db.query(models.CustomsBroker).filter(models.CustomsBroker.broker_key == broker_key).first()
        if not broker:
            return None
            
        new_personnel_data = personnel_data.model_dump()
        new_personnel = models.CustomsBrokerPersonnel(customs_broker_id=broker.id, **new_personnel_data)
        db.add(new_personnel)
        db.commit()
        db.refresh(new_personnel)
        return new_personnel

    @staticmethod
    def update_personnel(
        db: Session,
        broker_key: str,
        line: int,
        personnel_data: dto.CustomsBrokerPersonnelDTO,
    ):
        personnel = CustomsBrokerPersonnelService.get_by_broker_key_and_line(
            db, broker_key, line
        )
        if personnel:
            for key, value in personnel_data.model_dump(exclude_unset=True).items():
                setattr(personnel, key, value)
            db.commit()
            db.refresh(personnel)
        return personnel

    @staticmethod
    def delete_personnel(db: Session, broker_key: str, line: int):
        personnel = CustomsBrokerPersonnelService.get_by_broker_key_and_line(
            db, broker_key, line
        )
        if personnel:
            db.delete(personnel)
            db.commit()
        return personnel