feature/continuacion-vista-tablas-location

This commit is contained in:
hreyes
2026-03-12 16:14:34 -06:00
parent 678f91ecdc
commit 8d304fb98b
23 changed files with 1337 additions and 717 deletions

View File

@@ -1,3 +0,0 @@
"""
Módulo de localización
"""

View File

@@ -1,43 +0,0 @@
"""
DTOs (Data Transfer Objects) para módulo de localización
"""
from typing import Optional
from pydantic import BaseModel, Field
class LocationCreateDTO(BaseModel):
"""DTO para crear una localización"""
code: str = Field(..., max_length=5, description="Location code")
description: Optional[str] = Field(
None, max_length=200, description="Location description"
)
class Config:
from_attributes = True
class LocationUpdateDTO(BaseModel):
"""DTO para actualizar una localización"""
code: Optional[str] = Field(
None, max_length=5, description="Location code")
description: Optional[str] = Field(
None, max_length=200, description="Location description"
)
class Config:
from_attributes = True
class LocationResponseDTO(BaseModel):
"""DTO para responder con datos de una localización"""
id: int
code: str
description: Optional[str] = None
class Config:
from_attributes = True

View File

@@ -1,36 +0,0 @@
"""
Modelos ORM para gestión de localización
"""
from typing import Optional
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
from core.database import Base
from sqlalchemy import Integer, PrimaryKeyConstraint, String, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
class Location(Base, TenantScopedMixin, TimestampMixin):
"""
Modelo para la tabla Location - Localización
"""
__tablename__ = "location" # SLocalizacion
__table_args__ = (
PrimaryKeyConstraint("id", name="location_pkey"),
UniqueConstraint("code", name="location_code_unique"),
{"schema": "a24"},
)
# Primary key
id: Mapped[int] = mapped_column(
Integer, primary_key=True, autoincrement=True)
# Location code (unique)
code: Mapped[str] = mapped_column(String(5), nullable=False, unique=True)
# Location description
description: Mapped[Optional[str]] = mapped_column(String(200))
def __repr__(self):
return f"<Location(id={self.id}, code={self.code}, description={self.description})>"

View File

@@ -1,136 +0,0 @@
"""
Rutas para gestión de localización
"""
from typing import List
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session
from core.database import get_core_db
from .dto import LocationCreateDTO, LocationResponseDTO, LocationUpdateDTO
from .models import Location
from .service import LocationService
router = APIRouter(prefix="/locations", tags=["locations"])
@router.get(
"",
response_model=dict,
summary="Get all locations",
)
async def get_all_locations(
skip: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=100),
code: str = Query(None),
description: str = Query(None),
db: Session = Depends(get_core_db),
):
"""Get all locations with optional filtering and pagination"""
filters = {}
if code:
filters["code"] = code
if description:
filters["description"] = description
locations, total = LocationService.get_all(db, skip, limit, filters)
return {
"data": [LocationResponseDTO.model_validate(location) for location in locations],
"total": total,
"skip": skip,
"limit": limit,
}
@router.get(
"/{location_id}",
response_model=LocationResponseDTO,
summary="Get location by ID",
)
async def get_location(
location_id: int,
db: Session = Depends(get_core_db),
):
"""Get a location by its ID"""
location = LocationService.get_by_id(db, location_id)
if not location:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Location not found",
)
return LocationResponseDTO.model_validate(location)
@router.get(
"/code/{code}",
response_model=LocationResponseDTO,
summary="Get location by code",
)
async def get_location_by_code(
code: str,
db: Session = Depends(get_core_db),
):
"""Get a location by its code"""
location = LocationService.get_by_code(db, code)
if not location:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Location not found",
)
return LocationResponseDTO.model_validate(location)
@router.post(
"",
response_model=LocationResponseDTO,
status_code=status.HTTP_201_CREATED,
summary="Create location",
)
async def create_location(
location_data: LocationCreateDTO,
db: Session = Depends(get_core_db),
):
"""Create a new location"""
location = LocationService.create(db, location_data)
return LocationResponseDTO.model_validate(location)
@router.put(
"/{location_id}",
response_model=LocationResponseDTO,
summary="Update location",
)
async def update_location(
location_id: int,
location_data: LocationUpdateDTO,
db: Session = Depends(get_core_db),
):
"""Update a location"""
location = LocationService.update(db, location_id, location_data)
if not location:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Location not found",
)
return LocationResponseDTO.model_validate(location)
@router.delete(
"/{location_id}",
status_code=status.HTTP_204_NO_CONTENT,
summary="Delete location",
)
async def delete_location(
location_id: int,
db: Session = Depends(get_core_db),
):
"""Delete a location"""
success = LocationService.delete(db, location_id)
if not success:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Location not found",
)
return None

View File

@@ -1,136 +0,0 @@
"""
Capa de servicio para lógica de negocio de localización
"""
import logging
from typing import Any, Dict, List, Optional, Tuple
from fastapi import HTTPException
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from .dto import LocationCreateDTO, LocationResponseDTO, LocationUpdateDTO
from .models import Location
logger = logging.getLogger(__name__)
class LocationService:
"""Servicio para gestión de localización"""
def __init__(self, db: Session):
self.db = db
@staticmethod
def get_all(
db: Session,
skip: int = 0,
limit: int = 50,
filters: Optional[Dict[str, Any]] = None,
) -> Tuple[List[Location], int]:
"""Get all locations with pagination"""
query = db.query(Location)
if filters:
if filters.get("code"):
query = query.filter(
Location.code.ilike(f"%{filters['code']}%"))
if filters.get("description"):
query = query.filter(
Location.description.ilike(f"%{filters['description']}%")
)
total = query.count()
locations = query.offset(skip).limit(limit).all()
return locations, total
@staticmethod
def get_by_id(db: Session, location_id: int) -> Optional[Location]:
"""Get location by ID"""
return db.query(Location).filter(Location.id == location_id).first()
@staticmethod
def get_by_code(db: Session, code: str) -> Optional[Location]:
"""Get location by code"""
return db.query(Location).filter(Location.code == code).first()
@staticmethod
def create(db: Session, location_data: LocationCreateDTO) -> Location:
"""Create a new location"""
try:
db_location = Location(
**location_data.model_dump(exclude_unset=True))
db.add(db_location)
db.commit()
db.refresh(db_location)
return db_location
except IntegrityError as e:
db.rollback()
logger.error(f"IntegrityError creating location: {str(e)}")
raise HTTPException(
status_code=400,
detail="Location code already exists",
)
except Exception as e:
db.rollback()
logger.error(f"Error creating location: {str(e)}")
raise HTTPException(
status_code=500, detail="Error creating location")
@staticmethod
def update(
db: Session, location_id: int, location_data: LocationUpdateDTO
) -> Optional[Location]:
"""Update a location"""
try:
db_location = db.query(Location).filter(
Location.id == location_id).first()
if not db_location:
return None
for key, value in location_data.model_dump(exclude_unset=True).items():
setattr(db_location, key, value)
db.commit()
db.refresh(db_location)
return db_location
except IntegrityError as e:
db.rollback()
logger.error(f"IntegrityError updating location: {str(e)}")
raise HTTPException(
status_code=400,
detail="Error updating location",
)
except Exception as e:
db.rollback()
logger.error(f"Error updating location: {str(e)}")
raise HTTPException(
status_code=500, detail="Error updating location")
@staticmethod
def delete(db: Session, location_id: int) -> bool:
"""Delete a location"""
try:
db_location = db.query(Location).filter(
Location.id == location_id).first()
if not db_location:
return False
db.delete(db_location)
db.commit()
return True
except Exception as e:
db.rollback()
logger.error(f"Error deleting location: {str(e)}")
raise HTTPException(
status_code=500, detail="Error deleting location")

View File

@@ -0,0 +1 @@
# a76 general_catalogs.location

View File

@@ -0,0 +1,36 @@
from typing import Optional
from pydantic import BaseModel, ConfigDict, Field
from .models import LocationSystem
class LocationBase(BaseModel):
clave_localizacion: str = Field(
..., max_length=20, description="Clave/código de la localización"
)
localizacion: Optional[str] = Field(
None, max_length=200, description="Nombre o descripción"
)
system: LocationSystem = Field(
..., description="Contexto: fixed_asset (FA) o inventory"
)
class LocationCreate(LocationBase):
"""Optional extra fields for fixed_asset; used only when system == FIXED_ASSET."""
department: Optional[str] = Field(None, max_length=100)
responsible: Optional[str] = Field(None, max_length=200)
observations: Optional[str] = Field(None, description="Free text")
class LocationUpdate(BaseModel):
clave_localizacion: Optional[str] = Field(None, max_length=20)
localizacion: Optional[str] = Field(None, max_length=200)
system: Optional[LocationSystem] = None
class LocationResponse(LocationBase):
id: int
model_config = ConfigDict(from_attributes=True)

View File

@@ -0,0 +1,84 @@
"""
Modelo ORM para catálogo de localización (a76).
Tabla compartida para contexto Fixed Asset (FA) e inventory.
"""
import enum
from typing import Optional
from sqlalchemy import Integer, String, Text, UniqueConstraint, ForeignKey
from sqlalchemy.orm import Mapped, mapped_column
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
from core.database import Base
class LocationSystem(str, enum.Enum):
"""Contexto de uso de la localización."""
FIXED_ASSET = "fixed_asset"
INVENTORY = "inventory"
class Location(Base, TenantScopedMixin, TimestampMixin):
"""
Catálogo de localización (clave + localizacion), compartido por FA e inventory.
"""
__tablename__ = "location"
__table_args__ = (
UniqueConstraint(
"clave_localizacion",
"tenant_id",
"company_id",
"system",
name="uq_location_clave_tenant_company_system",
),
{"schema": "a76"},
)
id: Mapped[int] = mapped_column(
Integer, primary_key=True, autoincrement=True
)
clave_localizacion: Mapped[str] = mapped_column(String(20), nullable=False)
localizacion: Mapped[Optional[str]] = mapped_column(String(200), nullable=True)
system: Mapped[str] = mapped_column(
String(20), nullable=False
) # 'fixed_asset' | 'inventory'
def __repr__(self) -> str:
return (
f"<Location(id={self.id}, clave_localizacion={self.clave_localizacion!r}, "
f"localizacion={self.localizacion!r}, system={self.system!r})>"
)
class FaLocationExt(Base, TenantScopedMixin, TimestampMixin):
"""
Extra info for Fixed Asset locations only (1:1 with Location).
Table: Fa_Location_Ext (fa_location_ext).
"""
__tablename__ = "fa_location_ext"
__table_args__ = (
UniqueConstraint("location_id", name="uq_fa_location_ext_location_id"),
{"schema": "a76"},
)
id: Mapped[int] = mapped_column(
Integer, primary_key=True, autoincrement=True
)
location_id: Mapped[int] = mapped_column(
Integer,
ForeignKey("a76.location.id", ondelete="CASCADE"),
nullable=False,
unique=True,
)
department: Mapped[Optional[str]] = mapped_column(String(100), nullable=True)
responsible: Mapped[Optional[str]] = mapped_column(String(200), nullable=True)
observations: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
def __repr__(self) -> str:
return (
f"<FaLocationExt(id={self.id}, location_id={self.location_id}, "
f"department={self.department!r})>"
)

View File

@@ -0,0 +1,20 @@
from fastapi import APIRouter
from api.v1.common.tenant_crud_routes import TenantCRUDRoutes
from .dto import LocationCreate, LocationResponse, LocationUpdate
from .models import Location
from .service import LocationService
router = TenantCRUDRoutes(
service=LocationService,
create_schema=LocationCreate,
update_schema=LocationUpdate,
response_schema=LocationResponse,
prefix="/locations",
tags=["a76.general_catalogs.locations"],
resource_name="Location",
enable_list=True,
enable_filters=True,
max_page_size=1000,
).router

View File

@@ -0,0 +1,154 @@
from typing import Any, Dict, List, Optional, Tuple
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError
from .models import Location, FaLocationExt
from .dto import LocationCreate, LocationUpdate
class LocationService:
@staticmethod
def get_all(
db: Session,
tenant_id: int,
company_id: Optional[int],
skip: int = 0,
limit: int = 100,
filters: Optional[Dict[str, Any]] = None,
) -> Tuple[List[Location], int]:
query = db.query(Location).filter(Location.tenant_id == tenant_id)
if company_id is not None:
query = query.filter(Location.company_id == company_id)
if filters:
if filters.get("clave_localizacion"):
query = query.filter(
Location.clave_localizacion.ilike(
f"%{filters['clave_localizacion']}%"
)
)
if filters.get("localizacion"):
query = query.filter(
Location.localizacion.ilike(f"%{filters['localizacion']}%")
)
if filters.get("system"):
query = query.filter(Location.system == filters["system"])
total = query.count()
items = query.offset(skip).limit(limit).all()
return items, total
@staticmethod
def get_by_id(
db: Session,
id: int,
tenant_id: int,
company_id: int,
) -> Optional[Location]:
return (
db.query(Location)
.filter(
Location.id == id,
Location.tenant_id == tenant_id,
Location.company_id == company_id,
)
.first()
)
@staticmethod
def get_by_clave(
db: Session,
clave_localizacion: str,
tenant_id: int,
company_id: int,
system: Optional[str] = None,
) -> Optional[Location]:
query = db.query(Location).filter(
Location.clave_localizacion == clave_localizacion,
Location.tenant_id == tenant_id,
Location.company_id == company_id,
)
if system is not None:
query = query.filter(Location.system == system)
return query.first()
@staticmethod
def create(
db: Session,
data: LocationCreate,
tenant_id: int,
company_id: int,
) -> Location:
db_obj = Location(
clave_localizacion=data.clave_localizacion,
localizacion=data.localizacion,
system=data.system.value,
tenant_id=tenant_id,
company_id=company_id,
)
db.add(db_obj)
try:
db.flush() # get db_obj.id without committing
if data.system.value == "fixed_asset" and (
data.department is not None
or data.responsible is not None
or data.observations is not None
):
ext = FaLocationExt(
location_id=db_obj.id,
department=data.department,
responsible=data.responsible,
observations=data.observations,
tenant_id=tenant_id,
company_id=company_id,
)
db.add(ext)
db.commit()
db.refresh(db_obj)
return db_obj
except IntegrityError:
db.rollback()
raise ValueError("Ya existe una localización con esa clave y system")
@staticmethod
def update(
db: Session,
id: int,
tenant_id: int,
data: LocationUpdate,
company_id: int,
) -> Optional[Location]:
db_obj = LocationService.get_by_id(db, id, tenant_id, company_id)
if not db_obj:
return None
update_dict = data.model_dump(exclude_unset=True)
for key, value in update_dict.items():
if key == "system" and value is not None:
setattr(db_obj, key, value.value)
else:
setattr(db_obj, key, value)
try:
db.commit()
db.refresh(db_obj)
return db_obj
except IntegrityError:
db.rollback()
raise ValueError("Ya existe una localización con esa clave y system")
@staticmethod
def delete(
db: Session,
id: int,
tenant_id: int,
company_id: int,
) -> bool:
db_obj = LocationService.get_by_id(db, id, tenant_id, company_id)
if not db_obj:
return False
db.delete(db_obj)
db.commit()
return True

View File

@@ -25,12 +25,14 @@ from .error_catalogs.routes import router as error_catalogs_router
from .doda.routes import router as doda_router
from .prevalidators.routes import router as prevalidators_router
from .electronic_notices.routes import router as electronic_notices_router
from .location.routes import router as location_router
router = APIRouter()
router.include_router(company_router, tags=["a76 / company"])
router.include_router(package_router)
router.include_router(ports_router)
router.include_router(location_router)
router.include_router(tariff_fractions_router)
router.include_router(us_tariff_fractions_router)
router.include_router(historical_tariff_fractions_router, prefix="/fractions/historical-tariff-fractions", tags=["a76 / historical_tariff_fractions"])