Add service layers and models for Pedimento CRUD operations
- 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.
This commit is contained in:
140
backend/api/v1/modules/a76/pedmientos/services/pedimentos.py
Normal file
140
backend/api/v1/modules/a76/pedmientos/services/pedimentos.py
Normal file
@@ -0,0 +1,140 @@
|
||||
"""
|
||||
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
|
||||
Reference in New Issue
Block a user