ADUANASOFT/plantillas-proyectos
13
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'feature/items' (#56) from feature/items into development

Browse Source 
Reviewed-on: ADUANASOFT/anexo76#56
This commit is contained in:
acazares
2026-01-20 18:53:49 +00:00
parent 8a3d0eccec 7e9a1acd30
commit 0831dd0ab2
28 changed files with 1620 additions and 388 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
backend/api/v1/modules/a76/general_catalogs/tariff_fractions/dto.py
View File

@@ -38,11 +38,9 @@ class TariffFractionUpdateDTO(BaseModel):
class TariffFractionResponseDTO(BaseModel):
"""DTO para respuesta de fracción arancelaria"""
"""DTO para respuesta de fracción arancelaria (catálogo global)"""
id: int
tenant_id: int
company_id: int
code: str
fraction: str
description: Optional[str] = None
@@ -50,8 +48,6 @@ class TariffFractionResponseDTO(BaseModel):
umt: Optional[str] = None
adv_impo: Optional[str] = None
adv_expo: Optional[str] = None
created_at: datetime
updated_at: datetime
model_config = ConfigDict(from_attributes=True)

8
backend/api/v1/modules/a76/general_catalogs/tariff_fractions/models.py
View File

@@ -4,15 +4,15 @@ Modelos ORM para fracciones arancelarias (SITAR-SCAII)
from typing import Optional
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
from core.database import Base
from sqlalchemy import Integer, PrimaryKeyConstraint, String, Numeric
from sqlalchemy.orm import Mapped, mapped_column
class TariffFraction(Base, TenantScopedMixin, TimestampMixin):
class TariffFraction(Base):
"""
Modelo para fracciones arancelarias mexicanas (SITAR-SCAII)
Catálogo de referencia global (no tenant-scoped)
Corresponde a la tabla sFracciones
"""
@@ -22,10 +22,10 @@ class TariffFraction(Base, TenantScopedMixin, TimestampMixin):
{"schema": "a76", "extend_existing": True},
)
id: Mapped[int] = mapped_column(Integer, primary_key=True)
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
# Código completo de la fracción (ej: 01012101)
code: Mapped[str] = mapped_column(String(10), unique=True, index=True)
code: Mapped[str] = mapped_column(String(10), unique=True, index=True, nullable=False)
# Fracción formateada (ej: 0101.21.01)
fraction: Mapped[str] = mapped_column(String(15), index=True)

96
backend/api/v1/modules/a76/general_catalogs/tariff_fractions/routes.py
View File

@@ -1,5 +1,6 @@
"""
Endpoints API para fracciones arancelarias
Catálogo de referencia global (no tenant-scoped)
"""
from typing import Any, Dict, Optional
@@ -8,7 +9,6 @@ from sqlalchemy.orm import Session
from core.database import get_core_db
from core.security import get_current_user
from api.v1.common.tenant_crud_routes import TenantCRUDRoutes, validate_access_to_resource
from .dto import (
TariffFractionCreateDTO,
@@ -17,22 +17,6 @@ from .dto import (
)
from .service import TariffFractionService
# Create base router with generic CRUD routes (disabled list because we'll create a custom one)
base_router = TenantCRUDRoutes(
service=TariffFractionService,
create_schema=TariffFractionCreateDTO,
update_schema=TariffFractionUpdateDTO,
response_schema=TariffFractionResponseDTO,
prefix="/tariff-fractions",
tags=["a76 / general catalogs / tariff fractions"],
resource_name="TariffFraction",
id_name="tariff_fraction_id",
enable_list=False, # Disable default list, we'll add custom one
enable_filters=False,
default_page_size=50,
max_page_size=10000,
)
router = APIRouter(prefix="/tariff-fractions", tags=["a76 / general catalogs / tariff fractions"])
# Custom list endpoint with search filter
@@ -40,25 +24,22 @@ router = APIRouter(prefix="/tariff-fractions", tags=["a76 / general catalogs / t
"/",
response_model=Dict[str, Any],
summary="List Tariff Fractions",
description="Get paginated list of Tariff Fractions with optional search filter",
description="Get paginated list of Tariff Fractions with optional search filter (global catalog)",
)
async def list_tariff_fractions(
company_id: int = Query(..., description="Company ID"),
page: int = Query(1, ge=1, description="Page number"),
page_size: int = Query(50, ge=1, le=10000, description="Page size"),
search: Optional[str] = Query(None, description="Search in code, fraction, description, nico, or umt"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
tenant_id = validate_access_to_resource(db, company_id, current_user)
skip = (page - 1) * page_size
filters = {}
if search:
filters["search"] = search
items, total = TariffFractionService.get_all(
db, tenant_id, company_id, skip, page_size, filters
db, skip, page_size, filters
)
return {
@@ -69,6 +50,73 @@ async def list_tariff_fractions(
"pages": (total + page_size - 1) // page_size,
}
# Include other CRUD routes from base router
router.include_router(base_router.router)
@router.get(
"/{tariff_fraction_id}",
response_model=TariffFractionResponseDTO,
summary="Get Tariff Fraction by ID",
description="Get a specific tariff fraction by ID",
)
async def get_tariff_fraction(
tariff_fraction_id: int,
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
item = TariffFractionService.get_by_id(db, tariff_fraction_id)
if not item:
from fastapi import HTTPException
raise HTTPException(status_code=404, detail="Tariff fraction not found")
return TariffFractionResponseDTO.model_validate(item)
@router.post(
"/",
response_model=TariffFractionResponseDTO,
summary="Create Tariff Fraction",
description="Create a new tariff fraction (admin only)",
status_code=201,
)
async def create_tariff_fraction(
data: TariffFractionCreateDTO,
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
item = TariffFractionService.create(db, data)
return TariffFractionResponseDTO.model_validate(item)
@router.put(
"/{tariff_fraction_id}",
response_model=TariffFractionResponseDTO,
summary="Update Tariff Fraction",
description="Update an existing tariff fraction (admin only)",
)
async def update_tariff_fraction(
tariff_fraction_id: int,
data: TariffFractionUpdateDTO,
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
item = TariffFractionService.update(db, tariff_fraction_id, data)
if not item:
from fastapi import HTTPException
raise HTTPException(status_code=404, detail="Tariff fraction not found")
return TariffFractionResponseDTO.model_validate(item)
@router.delete(
"/{tariff_fraction_id}",
summary="Delete Tariff Fraction",
description="Delete a tariff fraction (admin only)",
status_code=204,
)
async def delete_tariff_fraction(
tariff_fraction_id: int,
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
success = TariffFractionService.delete(db, tariff_fraction_id)
if not success:
from fastapi import HTTPException
raise HTTPException(status_code=404, detail="Tariff fraction not found")

38
backend/api/v1/modules/a76/general_catalogs/tariff_fractions/service.py
View File

@@ -1,5 +1,6 @@
"""
Service para fracciones arancelarias
Catálogo de referencia global (no tenant-scoped)
"""
from typing import List, Optional, Tuple, Dict, Any
@@ -15,23 +16,18 @@ logger = logging.getLogger(__name__)
class TariffFractionService:
"""Service para gestionar fracciones arancelarias"""
"""Service para gestionar fracciones arancelarias (catálogo global)"""
@staticmethod
def get_all(
db: Session,
tenant_id: int,
company_id: int,
skip: int = 0,
limit: int = 100,
filters: Optional[Dict[str, Any]] = None,
) -> Tuple[List[TariffFraction], int]:
"""Obtiene todas las fracciones arancelarias con filtros opcionales"""
query = db.query(TariffFraction).filter(
TariffFraction.tenant_id == tenant_id,
TariffFraction.company_id == company_id,
)
query = db.query(TariffFraction)
# Aplicar filtros
if filters:
@@ -67,18 +63,12 @@ class TariffFractionService:
def get_by_id(
db: Session,
tariff_fraction_id: int,
tenant_id: int,
company_id: int,
) -> Optional[TariffFraction]:
"""Obtiene una fracción arancelaria por ID"""
return (
db.query(TariffFraction)
.filter(
TariffFraction.id == tariff_fraction_id,
TariffFraction.tenant_id == tenant_id,
TariffFraction.company_id == company_id,
)
.filter(TariffFraction.id == tariff_fraction_id)
.first()
)
@@ -86,18 +76,12 @@ class TariffFractionService:
def get_by_code(
db: Session,
code: str,
tenant_id: int,
company_id: int,
) -> Optional[TariffFraction]:
"""Obtiene una fracción arancelaria por código"""
return (
db.query(TariffFraction)
.filter(
TariffFraction.code == code,
TariffFraction.tenant_id == tenant_id,
TariffFraction.company_id == company_id,
)
.filter(TariffFraction.code == code)
.first()
)
@@ -105,16 +89,12 @@ class TariffFractionService:
def create(
db: Session,
tariff_fraction_data: TariffFractionCreateDTO,
tenant_id: int,
company_id: int,
) -> TariffFraction:
"""Crea una nueva fracción arancelaria"""
try:
tariff_fraction = TariffFraction(
**tariff_fraction_data.model_dump(),
tenant_id=tenant_id,
company_id=company_id,
)
db.add(tariff_fraction)
db.commit()
@@ -133,13 +113,11 @@ class TariffFractionService:
db: Session,
tariff_fraction_id: int,
tariff_fraction_data: TariffFractionUpdateDTO,
tenant_id: int,
company_id: int,
) -> Optional[TariffFraction]:
"""Actualiza una fracción arancelaria existente"""
tariff_fraction = TariffFractionService.get_by_id(
db, tariff_fraction_id, tenant_id, company_id
db, tariff_fraction_id
)
if not tariff_fraction:
@@ -165,13 +143,11 @@ class TariffFractionService:
def delete(
db: Session,
tariff_fraction_id: int,
tenant_id: int,
company_id: int,
) -> bool:
"""Elimina una fracción arancelaria"""
tariff_fraction = TariffFractionService.get_by_id(
db, tariff_fraction_id, tenant_id, company_id
db, tariff_fraction_id
)
if not tariff_fraction: