Merge pull request 'fixing/fixing_code' (#38) from fixing/fixing_code into development

Reviewed-on: ADUANASOFT/anexo76#38

backend/api/v1/modules/a76/general_catalogs/customs_broker_concepts/dto.py

@@ -1,27 +1,40 @@
+"""
+DTOs for Customs Broker Concepts.
+"""
+
+from datetime import datetime
 from typing import Optional
-from decimal import Decimal
+
 from pydantic import BaseModel, Field, ConfigDict
 
 
 class CustomsBrokerConceptBase(BaseModel):
     broker_key: str = Field(..., max_length=5, description="Customs Broker Key (CLAVEAA)")
-    concept: str = Field(..., max_length=15, description="Concept")
-    amount: Optional[Decimal] = Field(None, description="Amount")
-    priority: Optional[int] = Field(None, description="Priority")
+    concept: str = Field(..., max_length=15, description="Concept/Code")
+    amount: Optional[float] = Field(None, description="Amount (IMPORTE)")
+    priority: Optional[int] = Field(None, description="Priority (PRIORIDAD)")
 
 
 class CustomsBrokerConceptCreate(CustomsBrokerConceptBase):
+    """Schema for creating a concept"""
     pass
 
 
 class CustomsBrokerConceptUpdate(BaseModel):
-    broker_key: Optional[str] = Field(None, max_length=5)
-    concept: Optional[str] = Field(None, max_length=15)
-    amount: Optional[Decimal] = None
-    priority: Optional[int] = None
+    """Schema for updating a concept"""
+    broker_key: Optional[str] = Field(None, max_length=5, description="Customs Broker Key (CLAVEAA)")
+    concept: Optional[str] = Field(None, max_length=15, description="Concept/Code")
+    amount: Optional[float] = Field(None, description="Amount (IMPORTE)")
+    priority: Optional[int] = Field(None, description="Priority (PRIORIDAD)")
 
 
 class CustomsBrokerConceptResponse(CustomsBrokerConceptBase):
+    """Schema for concept response"""
     id: int
+    company_id: int
+    tenant_id: int
+    created_at: Optional[datetime] = None
+    updated_at: Optional[datetime] = None
 
     model_config = ConfigDict(from_attributes=True)
+

backend/api/v1/modules/a76/general_catalogs/customs_broker_concepts/routes.py

@@ -1,14 +1,26 @@
+"""
+Routes for managing Customs Broker Concepts.
+"""
+
 from api.v1.common.tenant_crud_routes import TenantCRUDRoutes
+
 from .dto import CustomsBrokerConceptCreate, CustomsBrokerConceptResponse, CustomsBrokerConceptUpdate
 from .service import CustomsBrokerConceptService
 
+# Create router using TenantCRUDRoutes factory
 router = TenantCRUDRoutes(
     service=CustomsBrokerConceptService,
     create_schema=CustomsBrokerConceptCreate,
     update_schema=CustomsBrokerConceptUpdate,
     response_schema=CustomsBrokerConceptResponse,
-    prefix="/customs-broker-concepts", 
-    tags=["a76.general_catalogs.customs_broker_concepts"],
+    prefix="/customs-broker-concepts",
+    tags=["a76 / customs_broker_concepts"],
     resource_name="Customs Broker Concept",
-    enable_list=True,
+    id_name="concept_id",
+    enable_list=True,  # Enable GET /customs-broker-concepts with pagination
+    enable_filters=True,  # Enable filtering
+    default_page_size=50,
+    max_page_size=100,
 ).router
+
+

backend/api/v1/modules/a76/general_catalogs/customs_broker_concepts/service.py

@@ -1,94 +1,125 @@
-from typing import List, Optional, Tuple, Dict, Any
+"""
+Service layer for Customs Broker Concepts.
+"""
+
+from typing import Optional, Tuple, List, Dict, Any
 import logging
 
 from sqlalchemy.orm import Session
-from sqlalchemy import select
 from sqlalchemy.exc import IntegrityError
 from fastapi import HTTPException
 
-from .models import CustomsBrokerConcept
-from .dto import CustomsBrokerConceptCreate, CustomsBrokerConceptUpdate
+from . import dto, models
 
 logger = logging.getLogger(__name__)
 
 
 class CustomsBrokerConceptService:
+    """Service for Customs Broker Concept CRUD operations with tenant support"""
+
     @staticmethod
     def get_all(
         db: Session,
         tenant_id: int,
         company_id: int,
         skip: int = 0,
-        limit: int = 100,
+        limit: int = 50,
         filters: Optional[Dict[str, Any]] = None,
-    ) -> Tuple[List[CustomsBrokerConcept], int]:
-        query = db.query(CustomsBrokerConcept).filter(
-            CustomsBrokerConcept.tenant_id == tenant_id,
-            CustomsBrokerConcept.company_id == company_id
+    ) -> Tuple[List[models.CustomsBrokerConcept], int]:
+        """Get all customs broker concepts for a tenant/company with pagination"""
+        query = db.query(models.CustomsBrokerConcept).filter(
+            models.CustomsBrokerConcept.tenant_id == tenant_id,
+            models.CustomsBrokerConcept.company_id == company_id,
         )
 
-        total = query.count()
-        items = query.offset(skip).limit(limit).all()
+        # Apply filters if provided
+        if filters:
+            if filters.get("broker_key"):
+                query = query.filter(
+                    models.CustomsBrokerConcept.broker_key.ilike(f"%{filters['broker_key']}%")
+                )
+            if filters.get("concept"):
+                query = query.filter(
+                    models.CustomsBrokerConcept.concept.ilike(
+                        f"%{filters['concept']}%")
+                )
 
-        return items, total
+        total = query.count()
+        concepts = query.offset(skip).limit(limit).all()
+
+        return concepts, total
 
     @staticmethod
     def get_by_id(
-        db: Session, id: int, tenant_id: int, company_id: int
-    ) -> Optional[CustomsBrokerConcept]:
-        return db.query(CustomsBrokerConcept).filter(
-            CustomsBrokerConcept.id == id,
-            CustomsBrokerConcept.tenant_id == tenant_id,
-            CustomsBrokerConcept.company_id == company_id
-        ).first()
+        db: Session, concept_id: int, tenant_id: int, company_id: int
+    ) -> Optional[models.CustomsBrokerConcept]:
+        """Get customs broker concept by ID"""
+        return (
+            db.query(models.CustomsBrokerConcept)
+            .filter(
+                models.CustomsBrokerConcept.id == concept_id,
+                models.CustomsBrokerConcept.tenant_id == tenant_id,
+                models.CustomsBrokerConcept.company_id == company_id,
+            )
+            .first()
+        )
 
     @staticmethod
     def create(
-        db: Session, data: CustomsBrokerConceptCreate, tenant_id: int, company_id: int
-    ) -> CustomsBrokerConcept:
-        db_obj = CustomsBrokerConcept(
-            **data.model_dump(),
-            tenant_id=tenant_id,
-            company_id=company_id
+        db: Session,
+        concept_data: dto.CustomsBrokerConceptCreate,
+        tenant_id: int,
+        company_id: int,
+    ) -> models.CustomsBrokerConcept:
+        """Create a new customs broker concept"""
+        new_concept = models.CustomsBrokerConcept(
+            **concept_data.model_dump(), tenant_id=tenant_id, company_id=company_id
         )
-        db.add(db_obj)
+        db.add(new_concept)
         db.commit()
-        db.refresh(db_obj)
-        return db_obj
+        db.refresh(new_concept)
+        return new_concept
 
     @staticmethod
     def update(
-        db: Session, id: int, tenant_id: int, data: CustomsBrokerConceptUpdate, company_id: int
-    ) -> Optional[CustomsBrokerConcept]:
-        db_obj = CustomsBrokerConceptService.get_by_id(
-            db, id, tenant_id, company_id)
-        if not db_obj:
+        db: Session,
+        concept_id: int,
+        tenant_id: int,
+        concept_data: dto.CustomsBrokerConceptUpdate,
+        company_id: int,
+    ) -> Optional[models.CustomsBrokerConcept]:
+        """Update a customs broker concept"""
+        concept = CustomsBrokerConceptService.get_by_id(
+            db, concept_id, tenant_id, company_id)
+        if not concept:
             return None
 
-        update_dict = data.model_dump(exclude_unset=True)
-        for key, value in update_dict.items():
-            setattr(db_obj, key, value)
+        # Update fields
+        update_data = concept_data.model_dump(exclude_unset=True)
+        for field, value in update_data.items():
+            setattr(concept, field, value)
 
         db.commit()
-        db.refresh(db_obj)
-        return db_obj
+        db.refresh(concept)
+        return concept
 
     @staticmethod
     def delete(
-        db: Session, id: int, tenant_id: int, company_id: int
+        db: Session, concept_id: int, tenant_id: int, company_id: int
     ) -> bool:
-        db_obj = CustomsBrokerConceptService.get_by_id(
-            db, id, tenant_id, company_id)
-        if not db_obj:
+        """Delete a customs broker concept"""
+        concept = CustomsBrokerConceptService.get_by_id(
+            db, concept_id, tenant_id, company_id)
+        if not concept:
             return False
 
         try:
-            db.delete(db_obj)
+            db.delete(concept)
             db.commit()
             return True
         except IntegrityError as e:
             db.rollback()
-            logger.error(f"IntegrityError deleting customs broker concept {id}: {str(e)}")
+            logger.error(f"IntegrityError deleting concept {concept_id}: {str(e)}")
             if "foreign key constraint" in str(e).lower():
                 raise HTTPException(
                     status_code=400, 
@@ -97,5 +128,5 @@ class CustomsBrokerConceptService:
             raise HTTPException(status_code=400, detail="Error al eliminar el concepto")
         except Exception as e:
             db.rollback()
-            logger.error(f"Error deleting customs broker concept {id}: {str(e)}")
+            logger.error(f"Error deleting concept {concept_id}: {str(e)}")
             raise HTTPException(status_code=500, detail="Error al eliminar el concepto")