feat: Implement Fixed Asset (FA) line item management

- Added new schemas for FA line items in the backend, including creation, update, and response DTOs.
- Updated existing line item schemas to include FA data.
- Modified database models to reflect new table names for line quantities and references.
- Enhanced ItemService to handle FA data during item creation and updates.
- Introduced new routes and service layer for FA line items, including CRUD operations.
- Updated frontend components to support FA line item data, including new fields and UI adjustments.
- Implemented data flattening for improved item display in the dashboard.
This commit is contained in:
2026-01-12 12:23:40 -06:00
parent 6845dcd670
commit 8dab1d6e46
19 changed files with 852 additions and 173 deletions

View File

@@ -0,0 +1,17 @@
"""
Módulo de líneas de activos fijos (FA Item Lines) para Anexo 24
"""
from .models import FaLineItem
from .dto import FaLineItemCreateDTO, FaLineItemUpdateDTO, FaLineItemResponseDTO
from .service import FaLineItemService
from .routes import router
__all__ = [
"FaLineItem",
"FaLineItemCreateDTO",
"FaLineItemUpdateDTO",
"FaLineItemResponseDTO",
"FaLineItemService",
"router",
]

View File

@@ -0,0 +1,167 @@
"""
DTOs (Data Transfer Objects) para líneas de activos fijos (FA Item Lines)
"""
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, ConfigDict, Field
class FaLineItemCreateDTO(BaseModel):
"""DTO para crear una línea de activo fijo"""
# line_item_id references the id in a76.item_lines
line_item_id: int = Field(..., description="ID de la línea base en a76.item_lines")
# Asset information (SCAF specific)
asset_number: Optional[str] = Field(
None, max_length=25, description="Número de activo"
)
asset_photo: Optional[str] = Field(
None, max_length=255, description="Foto del activo fijo"
)
equipment_message: Optional[str] = Field(
None, max_length=40, description="Mensaje del equipo"
)
invoice_type_asset: Optional[str] = Field(
None, max_length=6, description="Tipo de factura para activo"
)
return_import_invoice: Optional[str] = Field(
None, max_length=15, description="Factura de importación de retorno"
)
return_import_date: Optional[int] = Field(
None, description="Fecha de factura de importación de retorno"
)
movement_type_import: Optional[str] = Field(
None, max_length=3, description="Tipo de movimiento de importación"
)
# Cross-references for import repair
search_invoice: Optional[str] = Field(
None, max_length=15, description="Factura de búsqueda cruzada"
)
search_line: Optional[int] = Field(None, description="Línea de búsqueda cruzada")
# Search type
search_type: Optional[str] = Field(
None, max_length=10, description="Tipo de búsqueda"
)
# Subitems
is_subitem: Optional[bool] = Field(False, description="Es subpartida")
contains_subitems: Optional[bool] = Field(False, description="Contiene subpartidas")
includes_subitems: Optional[bool] = Field(False, description="Incluye subpartidas")
subitem_number: Optional[int] = Field(None, description="Número de subpartida")
# Special flags
download: Optional[bool] = Field(None, description="Indicador de descarga")
own_equipment: Optional[bool] = Field(None, description="Equipo propio")
omit_annex31: Optional[bool] = Field(None, description="Omitir en Anexo 31")
class Config:
from_attributes = True
class FaLineItemUpdateDTO(BaseModel):
"""DTO para actualizar una línea de activo fijo"""
# Asset information (SCAF specific)
asset_number: Optional[str] = Field(
None, max_length=25, description="Número de activo"
)
asset_photo: Optional[str] = Field(
None, max_length=255, description="Foto del activo fijo"
)
equipment_message: Optional[str] = Field(
None, max_length=40, description="Mensaje del equipo"
)
invoice_type_asset: Optional[str] = Field(
None, max_length=6, description="Tipo de factura para activo"
)
return_import_invoice: Optional[str] = Field(
None, max_length=15, description="Factura de importación de retorno"
)
return_import_date: Optional[int] = Field(
None, description="Fecha de factura de importación de retorno"
)
movement_type_import: Optional[str] = Field(
None, max_length=3, description="Tipo de movimiento de importación"
)
# Cross-references for import repair
search_invoice: Optional[str] = Field(
None, max_length=15, description="Factura de búsqueda cruzada"
)
search_line: Optional[int] = Field(None, description="Línea de búsqueda cruzada")
# Search type
search_type: Optional[str] = Field(
None, max_length=10, description="Tipo de búsqueda"
)
# Subitems
is_subitem: Optional[bool] = Field(False, description="Es subpartida")
contains_subitems: Optional[bool] = Field(False, description="Contiene subpartidas")
includes_subitems: Optional[bool] = Field(False, description="Incluye subpartidas")
subitem_number: Optional[int] = Field(None, description="Número de subpartida")
# Special flags
download: Optional[bool] = Field(None, description="Indicador de descarga")
own_equipment: Optional[bool] = Field(None, description="Equipo propio")
omit_annex31: Optional[bool] = Field(None, description="Omitir en Anexo 31")
class Config:
from_attributes = True
class FaLineItemResponseDTO(BaseModel):
"""DTO para respuesta de línea de activo fijo"""
id: int = Field(..., description="ID de la línea de activo fijo")
tenant_id: int = Field(..., description="ID del tenant")
company_id: int = Field(..., description="ID de la empresa")
# Asset information (SCAF specific)
asset_number: Optional[str] = Field(None, description="Número de activo")
asset_photo: Optional[str] = Field(None, description="Foto del activo fijo")
equipment_message: Optional[str] = Field(None, description="Mensaje del equipo")
invoice_type_asset: Optional[str] = Field(
None, description="Tipo de factura para activo"
)
return_import_invoice: Optional[str] = Field(
None, description="Factura de importación de retorno"
)
return_import_date: Optional[int] = Field(
None, description="Fecha de factura de importación de retorno"
)
movement_type_import: Optional[str] = Field(
None, description="Tipo de movimiento de importación"
)
# Cross-references for import repair
search_invoice: Optional[str] = Field(
None, description="Factura de búsqueda cruzada"
)
search_line: Optional[int] = Field(None, description="Línea de búsqueda cruzada")
# Search type
search_type: Optional[str] = Field(None, description="Tipo de búsqueda")
# Subitems
is_subitem: Optional[bool] = Field(False, description="Es subpartida")
contains_subitems: Optional[bool] = Field(False, description="Contiene subpartidas")
includes_subitems: Optional[bool] = Field(False, description="Incluye subpartidas")
subitem_number: Optional[int] = Field(None, description="Número de subpartida")
# Special flags
download: Optional[bool] = Field(None, description="Indicador de descarga")
own_equipment: Optional[bool] = Field(None, description="Equipo propio")
omit_annex31: Optional[bool] = Field(None, description="Omitir en Anexo 31")
# Timestamps
created_at: datetime = Field(..., description="Fecha de creación")
updated_at: datetime = Field(..., description="Fecha de actualización")
class Config:
from_attributes = True

View File

@@ -1,34 +1,84 @@
from typing import Optional
from sqlalchemy import Boolean, ForeignKey, Integer, String
from sqlalchemy.orm import Mapped, mapped_column
"""
Modelo ORM para datos específicos de líneas de Activos Fijos - Anexo 24
Extensión de a76.item_lines para SCAF (Sistema de Control de Activo Fijo)
"""
from typing import TYPE_CHECKING, Optional
from sqlalchemy import (
Boolean,
ForeignKeyConstraint,
Integer,
PrimaryKeyConstraint,
String,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
from core.database import Base
class LineItem(Base):
__tablename__ = "line_items"
if TYPE_CHECKING:
from api.v1.modules.a76.items.line_items.models import LineItem
class FaLineItem(Base, TenantScopedMixin, TimestampMixin):
"""
Tabla fa_item_lines: Extensión de Anexo 24 para líneas de Activos Fijos.
Hereda el ID de la tabla item_lines en a76.
"""
__tablename__ = "fa_item_lines"
__table_args__ = (
{"schema": "a24"}
PrimaryKeyConstraint("id", name="fa_item_lines_pkey"),
ForeignKeyConstraint(
["id"], ["a76.item_lines.id"], name="fk_fa_item_lines_master"
),
{"schema": "a24"},
)
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
item_id: Mapped[int] = mapped_column(ForeignKey("a76.items.id"))
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=False)
# Asset information (SCAF specific)
asset_number: Mapped[Optional[str]] = mapped_column(String(25)) # ASSETNUMBER
asset_photo: Mapped[Optional[str]] = mapped_column(String(255)) # FOTOACTIVOFIJO
equipment_message: Mapped[Optional[str]] = mapped_column(String(40)) # EQI_MENSAJE
invoice_type_asset: Mapped[Optional[str]] = mapped_column(String(6)) # TIPOFACTURAASSET
return_import_invoice: Mapped[Optional[str]] = mapped_column(String(15)) # FACTURAIMPORET
return_import_date: Mapped[Optional[int]] = mapped_column(Integer) # FECHAFACIMPORET
movement_type_import: Mapped[Optional[str]] = mapped_column(String(3)) # TIPOMOVIMPO
equipment_message: Mapped[Optional[str]] = mapped_column(String(40)) # EQI_MENSAJE
invoice_type_asset: Mapped[Optional[str]] = mapped_column(
String(6)
) # TIPOFACTURAASSET
return_import_invoice: Mapped[Optional[str]] = mapped_column(
String(15)
) # FACTURAIMPORET
return_import_date: Mapped[Optional[int]] = mapped_column(
Integer
) # FECHAFACIMPORET
movement_type_import: Mapped[Optional[str]] = mapped_column(
String(3)
) # TIPOMOVIMPO
# Cross-references IN CASE OF IMPORT REPAIR
search_invoice: Mapped[Optional[str]] = mapped_column(String(15)) # FACTURAEXPO (when line is import) / FACTURAIMPO (when line is export)
search_line: Mapped[Optional[int]] = mapped_column(Integer) # LINEAEXPO (when line is import) / LINEAIMPO (when line is export)
search_invoice: Mapped[Optional[str]] = mapped_column(
String(15)
) # FACTURAEXPO (when line is import) / FACTURAIMPO (when line is export)
search_line: Mapped[Optional[int]] = mapped_column(
Integer
) # LINEAEXPO (when line is import) / LINEAIMPO (when line is export)
# Search type
search_type: Mapped[Optional[str]] = mapped_column(String(10)) # TIPOBUSQUEDA
# Special flags
download: Mapped[Optional[bool]] = mapped_column(Boolean) # DESCARGA
# Subitems
is_subitem: Mapped[Optional[bool]] = mapped_column(Boolean) # ESSUBPARTIDA
contains_subitems: Mapped[Optional[bool]] = mapped_column(Boolean) # CONTIENESUBP
includes_subitems: Mapped[Optional[bool]] = mapped_column(
Boolean
) # INCUYESUBPARTIDAS
subitem_number: Mapped[Optional[int]] = mapped_column(Integer) # SUBPARTIDA
# Special flags
download: Mapped[Optional[bool]] = mapped_column(Boolean) # DESCARGA
own_equipment: Mapped[Optional[bool]] = mapped_column(Boolean) # EQUIPOPROPIO
omit_annex31: Mapped[Optional[bool]] = mapped_column(Boolean) # OMITITENANEXO31
omit_annex31: Mapped[Optional[bool]] = mapped_column(Boolean) # OMITITENANEXO31
# --- RELACIÓN ---
master_info: Mapped["LineItem"] = relationship("LineItem", back_populates="fa_data")
def __repr__(self) -> str:
return f"<FaLineItem(id={self.id}, asset_number='{self.asset_number}')>"

View File

@@ -0,0 +1,35 @@
"""
Endpoints API para gestión de líneas de activos fijos (FA Item Lines)
"""
from typing import Any, Dict
from fastapi import Depends, Query
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 FaLineItemCreateDTO, FaLineItemResponseDTO, FaLineItemUpdateDTO
from .service import FaLineItemService
# Create router with generic CRUD routes
crud_routes = TenantCRUDRoutes(
service=FaLineItemService,
create_schema=FaLineItemCreateDTO,
update_schema=FaLineItemUpdateDTO,
response_schema=FaLineItemResponseDTO,
prefix="/fa/item-lines",
tags=["a24 / fa / item-lines"],
resource_name="Fixed Asset Line Item",
id_name="fa_line_item_id",
enable_list=True,
enable_filters=True,
default_page_size=50,
max_page_size=100,
)
router = crud_routes.router

View File

@@ -0,0 +1,258 @@
"""
Capa de servicio para lógica de negocio de líneas de activos fijos (FA Item Lines)
"""
import logging
from typing import Any, Dict, List, Optional
from fastapi import HTTPException
from sqlalchemy import and_
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from .dto import FaLineItemCreateDTO, FaLineItemResponseDTO, FaLineItemUpdateDTO
from .models import FaLineItem
logger = logging.getLogger(__name__)
class FaLineItemService:
"""Servicio para gestión de líneas de activos fijos"""
@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[FaLineItem], int]:
"""
Obtener todas las líneas de activos fijos con paginación y filtros
"""
query = db.query(FaLineItem).filter(
FaLineItem.tenant_id == tenant_id, FaLineItem.company_id == company_id
)
if filters:
if filters.get("asset_number"):
query = query.filter(
FaLineItem.asset_number.ilike(f"%{filters['asset_number']}%")
)
if filters.get("invoice_type_asset"):
query = query.filter(
FaLineItem.invoice_type_asset == filters["invoice_type_asset"]
)
if filters.get("own_equipment") is not None:
query = query.filter(
FaLineItem.own_equipment == filters["own_equipment"]
)
if filters.get("download") is not None:
query = query.filter(FaLineItem.download == filters["download"])
total = query.count()
items = query.offset(skip).limit(limit).all()
return items, total
@staticmethod
def get_by_id(
db: Session, fa_line_item_id: int, tenant_id: int, company_id: int
) -> Optional[FaLineItem]:
"""Obtener una línea de activo fijo por ID"""
return (
db.query(FaLineItem)
.filter(
FaLineItem.id == fa_line_item_id,
FaLineItem.tenant_id == tenant_id,
FaLineItem.company_id == company_id,
)
.first()
)
@staticmethod
def get_by_line_item_id(
db: Session, line_item_id: int, tenant_id: int, company_id: int
) -> Optional[FaLineItem]:
"""Obtener una línea de activo fijo por line_item_id de a76"""
return (
db.query(FaLineItem)
.filter(
FaLineItem.id == line_item_id,
FaLineItem.tenant_id == tenant_id,
FaLineItem.company_id == company_id,
)
.first()
)
@staticmethod
def create(
db: Session,
fa_line_item_data: FaLineItemCreateDTO,
tenant_id: int,
company_id: int,
) -> FaLineItem:
"""Crear una nueva línea de activo fijo"""
try:
# Verificar que la línea base existe en a76.item_lines
from api.v1.modules.a76.items.line_items.models import LineItem
base_line_item = (
db.query(LineItem)
.filter(
LineItem.id == fa_line_item_data.line_item_id,
LineItem.tenant_id == tenant_id,
LineItem.company_id == company_id,
)
.first()
)
if not base_line_item:
raise HTTPException(
status_code=404,
detail=f"Base line item with id {fa_line_item_data.line_item_id} not found",
)
# Verificar si ya existe una línea FA para esta línea base
existing_fa = (
db.query(FaLineItem)
.filter(
FaLineItem.id == fa_line_item_data.line_item_id,
FaLineItem.tenant_id == tenant_id,
FaLineItem.company_id == company_id,
)
.first()
)
if existing_fa:
raise HTTPException(
status_code=400,
detail=f"FA line item already exists for line_item_id {fa_line_item_data.line_item_id}",
)
# Crear nueva línea FA
fa_line_item = FaLineItem(
id=fa_line_item_data.line_item_id, # Usa el mismo ID que la línea base
tenant_id=tenant_id,
company_id=company_id,
asset_number=fa_line_item_data.asset_number,
asset_photo=fa_line_item_data.asset_photo,
equipment_message=fa_line_item_data.equipment_message,
invoice_type_asset=fa_line_item_data.invoice_type_asset,
return_import_invoice=fa_line_item_data.return_import_invoice,
return_import_date=fa_line_item_data.return_import_date,
movement_type_import=fa_line_item_data.movement_type_import,
search_invoice=fa_line_item_data.search_invoice,
search_line=fa_line_item_data.search_line,
search_type=fa_line_item_data.search_type,
download=fa_line_item_data.download,
own_equipment=fa_line_item_data.own_equipment,
omit_annex31=fa_line_item_data.omit_annex31,
)
db.add(fa_line_item)
db.commit()
db.refresh(fa_line_item)
logger.info(
f"FA line item created: id={fa_line_item.id}, tenant={tenant_id}, company={company_id}"
)
return fa_line_item
except IntegrityError as e:
db.rollback()
logger.error(f"IntegrityError creating FA line item: {str(e)}")
raise HTTPException(
status_code=400,
detail="Error creating FA line item. Constraint violation.",
)
except HTTPException:
db.rollback()
raise
except Exception as e:
db.rollback()
logger.error(f"Error creating FA line item: {str(e)}")
raise HTTPException(
status_code=500, detail="Internal server error creating FA line item"
)
@staticmethod
def update(
db: Session,
fa_line_item_id: int,
fa_line_item_data: FaLineItemUpdateDTO,
tenant_id: int,
company_id: int,
) -> FaLineItem:
"""Actualizar una línea de activo fijo existente"""
try:
fa_line_item = FaLineItemService.get_by_id(
db, fa_line_item_id, tenant_id, company_id
)
if not fa_line_item:
raise HTTPException(
status_code=404,
detail=f"FA line item with id {fa_line_item_id} not found",
)
# Actualizar solo los campos proporcionados
update_data = fa_line_item_data.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(fa_line_item, field, value)
db.commit()
db.refresh(fa_line_item)
logger.info(
f"FA line item updated: id={fa_line_item.id}, tenant={tenant_id}, company={company_id}"
)
return fa_line_item
except HTTPException:
db.rollback()
raise
except Exception as e:
db.rollback()
logger.error(f"Error updating FA line item: {str(e)}")
raise HTTPException(
status_code=500, detail="Internal server error updating FA line item"
)
@staticmethod
def delete(
db: Session, fa_line_item_id: int, tenant_id: int, company_id: int
) -> bool:
"""Eliminar una línea de activo fijo"""
try:
fa_line_item = FaLineItemService.get_by_id(
db, fa_line_item_id, tenant_id, company_id
)
if not fa_line_item:
raise HTTPException(
status_code=404,
detail=f"FA line item with id {fa_line_item_id} not found",
)
db.delete(fa_line_item)
db.commit()
logger.info(
f"FA line item deleted: id={fa_line_item_id}, tenant={tenant_id}, company={company_id}"
)
return True
except HTTPException:
db.rollback()
raise
except Exception as e:
db.rollback()
logger.error(f"Error deleting FA line item: {str(e)}")
raise HTTPException(
status_code=500, detail="Internal server error deleting FA line item"
)

View File

@@ -6,9 +6,13 @@ from fastapi import APIRouter
# Importar routers de submódulos
from .fa.fa_classes.routes import router as fa_classes_router
from .fa.fa_item_lines.routes import router as fa_item_lines_router
# Router principal de A24
router = APIRouter()
# Registrar routers de FA (Fixed Assets)
router.include_router(fa_classes_router, prefix="/a24", tags=["a24 / fa / classes"])
router.include_router(
fa_item_lines_router, prefix="/a24", tags=["a24 / fa / item-lines"]
)