acazares
07dfe1edb1
- Implemented service classes for PedimentoConfigParameters, PedimentoConfigSurcharges, PedimentoConfigUpdateRectification, PedimentoConfigUpdates, PedimentoCustomsOffices, PedimentoDates, PedimentoDecrementables, PedimentoIncrementables, PedimentoIndexes, PedimentoPayments, PedimentoRectificationDestination, PedimentoRectificationOrigin, PedimentoTransportMeans, and PedimentoValidation. - Each service class includes methods for CRUD operations: create, read, update, and delete. - Added a main router for the API v1, integrating various modules including authentication, tenants, licenses, and pedimentos. - Created models for PedimentoValidation with appropriate constraints and relationships.
141 lines
4.0 KiB
Python
141 lines
4.0 KiB
Python
"""
|
|
Service layer for Pedimentos CRUD operations
|
|
"""
|
|
from typing import List, Optional, Dict, Any
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import desc
|
|
from fastapi import HTTPException
|
|
|
|
from ..models.pedimentos import Pedimentos
|
|
from ..dtos.pedimentos import PedimentosCreate, PedimentosUpdate
|
|
|
|
|
|
class PedimentosService:
|
|
"""Service class for Pedimentos business logic"""
|
|
|
|
@staticmethod
|
|
def get_all(
|
|
db: Session,
|
|
tenant_id: int,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
filters: Optional[Dict[str, Any]] = None
|
|
) -> tuple[List[Pedimentos], int]:
|
|
"""
|
|
Get all pedimentos for a tenant with pagination and filters
|
|
|
|
Args:
|
|
db: Database session
|
|
tenant_id: Tenant ID
|
|
skip: Number of records to skip
|
|
limit: Maximum number of records to return
|
|
filters: Optional filters dict
|
|
|
|
Returns:
|
|
Tuple of (list of pedimentos, total count)
|
|
"""
|
|
query = db.query(Pedimentos).filter(Pedimentos.tenant_id == tenant_id)
|
|
|
|
if filters:
|
|
if filters.get("status"):
|
|
query = query.filter(Pedimentos.status == filters["status"])
|
|
if filters.get("client_id"):
|
|
query = query.filter(Pedimentos.client_id == filters["client_id"])
|
|
if filters.get("year"):
|
|
query = query.filter(Pedimentos.year == filters["year"])
|
|
|
|
total = query.count()
|
|
items = query.order_by(desc(Pedimentos.created_at)).offset(skip).limit(limit).all()
|
|
|
|
return items, total
|
|
|
|
@staticmethod
|
|
def get_by_id(db: Session, pedimento_id: int, tenant_id: int) -> Optional[Pedimentos]:
|
|
"""
|
|
Get a pedimento by ID
|
|
|
|
Args:
|
|
db: Database session
|
|
pedimento_id: Pedimento ID
|
|
tenant_id: Tenant ID
|
|
|
|
Returns:
|
|
Pedimento or None if not found
|
|
"""
|
|
return db.query(Pedimentos).filter(
|
|
Pedimentos.id == pedimento_id,
|
|
Pedimentos.tenant_id == tenant_id
|
|
).first()
|
|
|
|
@staticmethod
|
|
def create(db: Session, pedimento_data: PedimentosCreate, tenant_id: int) -> Pedimentos:
|
|
"""
|
|
Create a new pedimento
|
|
|
|
Args:
|
|
db: Database session
|
|
pedimento_data: Pedimento creation data
|
|
|
|
Returns:
|
|
Created pedimento
|
|
"""
|
|
pedimento = Pedimentos(**pedimento_data.model_dump())
|
|
pedimento.tenant_id = 1
|
|
|
|
db.add(pedimento)
|
|
db.commit()
|
|
db.refresh(pedimento)
|
|
return pedimento
|
|
|
|
@staticmethod
|
|
def update(
|
|
db: Session,
|
|
pedimento_id: int,
|
|
tenant_id: int,
|
|
pedimento_data: PedimentosUpdate
|
|
) -> Optional[Pedimentos]:
|
|
"""
|
|
Update a pedimento
|
|
|
|
Args:
|
|
db: Database session
|
|
pedimento_id: Pedimento ID
|
|
tenant_id: Tenant ID
|
|
pedimento_data: Updated data
|
|
|
|
Returns:
|
|
Updated pedimento or None if not found
|
|
"""
|
|
pedimento = PedimentosService.get_by_id(db, pedimento_id, tenant_id)
|
|
if not pedimento:
|
|
return None
|
|
|
|
update_data = pedimento_data.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(pedimento, field, value)
|
|
|
|
db.commit()
|
|
db.refresh(pedimento)
|
|
return pedimento
|
|
|
|
@staticmethod
|
|
def delete(db: Session, pedimento_id: int, tenant_id: int) -> bool:
|
|
"""
|
|
Delete a pedimento
|
|
|
|
Args:
|
|
db: Database session
|
|
pedimento_id: Pedimento ID
|
|
tenant_id: Tenant ID
|
|
|
|
Returns:
|
|
True if deleted, False if not found
|
|
"""
|
|
pedimento = PedimentosService.get_by_id(db, pedimento_id, tenant_id)
|
|
if not pedimento:
|
|
return False
|
|
|
|
db.delete(pedimento)
|
|
db.commit()
|
|
return True
|