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

TODOS de configuracion general lista

Browse Source
This commit is contained in:
Kevin_Ramirez
2026-04-13 18:02:39 -05:00
parent 1d53ca9562
commit e5fa5f0326
49 changed files with 1678 additions and 1233 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
backend/api/v1/modules/a76/items/exports/validators/common.py
View File

@@ -1,6 +1,6 @@
from sqlalchemy import exists
from sqlalchemy.orm import Session
from api.v1.modules.a76.invoices.common.common_validators import invoice_exists_by_id
from api.v1.modules.a76.invoices.common.common_validators import invoice_id_required
from core.exceptions import ErrorCollector
from sqlalchemy import func
@@ -32,7 +32,7 @@ def validate_common(
errors: ErrorCollector,
line_number: int,
):
invoice: InvoiceHeader = invoice_exists_by_id(
invoice: InvoiceHeader = invoice_id_required(
db, line.invoice_id, tenant_id, company_id, errors
)
line_item: LineItem = item_exists(db, line.line_number, tenant_id, company_id)

20
backend/api/v1/modules/a76/items/exports/validators/create.py
View File

@@ -53,8 +53,16 @@ def validate_create(
if not line.class_id:
errors.add_required_error(field=f"line[{line_number}].class_id")
if not line.quantity or not line.quantity.quantity or line.quantity.quantity <= 0:
if not line.quantity or line.quantity.quantity is None:
errors.add_required_error(field=f"line[{line_number}].quantity.quantity")
elif line.quantity.quantity <= 0:
errors.add_error(
field=f"line[{line_number}].quantity.quantity",
message=f"La cantidad debe ser mayor a cero (recibido: {line.quantity.quantity})",
solution=["Capturar una cantidad válida mayor a cero."],
code="INVALID_QUANTITY",
value=float(line.quantity.quantity)
)
# TODO: Añadir validacion SSisGen:CalcularCostoUnitarioEnBaseAValorTotalScaf <-- de la tabla de preferencias de el sistema
# if SSisGen:CalcularCostoUnitarioEnBaseAValorTotalScaf == False:
@@ -68,8 +76,16 @@ def validate_create(
field=f"line[{line_number}].financial.unit_cost_capture"
)
if not line.quantity or not line.quantity.net_weight or line.quantity.net_weight <= 0:
if not line.quantity or line.quantity.net_weight is None:
errors.add_required_error(field=f"line[{line_number}].quantity.net_weight")
elif line.quantity.net_weight <= 0:
errors.add_error(
field=f"line[{line_number}].quantity.net_weight",
message=f"El peso neto debe ser mayor a cero (recibido: {line.quantity.net_weight})",
solution=["Capturar un peso neto válido mayor a cero."],
code="INVALID_NET_WEIGHT",
value=float(line.quantity.net_weight)
)
if not line.customs or not line.customs.origin_country:
errors.add_required_error(field=f"line[{line_number}].customs.origin_country")

10
backend/api/v1/modules/a76/items/exports/validators/update.py
View File

@@ -184,28 +184,28 @@ def validate_update(
line.order = existing_line.order
# Descripciones
if not line.description.description_spanish:
if line.description.description_spanish is None:
line.description.description_spanish = (
existing_line.description.description_spanish
)
if not line.description.description_english:
if line.description.description_english is None:
line.description.description_english = (
existing_line.description.description_english
)
if not line.description.extra_description:
if line.description.extra_description is None:
line.description.extra_description = (
existing_line.description.extra_description
)
# Marca y modelo
if line.description.brand:
if line.description.brand is not None:
line.description.brand = line.description.brand.upper().strip()
else:
line.description.brand = existing_line.description.brand
if line.description.model:
if line.description.model is not None:
line.description.model = line.description.model.upper().strip()
else:
line.description.model = existing_line.description.model

5
backend/api/v1/modules/a76/items/imports/validators/calculations.py
View File

@@ -81,6 +81,11 @@ def calculate_values(
return
currency, currency_type, exchange_rate = result
# Safety guard: Ensure nested objects exist before calculating
if not line.financial or not line.quantity:
return
# Prioridad: currency_type para alinear con create.py y CSV
if currency_type in ("USD", "ME"):
currency = "foreign"

4
backend/api/v1/modules/a76/items/imports/validators/common.py
View File

@@ -1,6 +1,6 @@
from sqlalchemy import exists
from sqlalchemy.orm import Session
from api.v1.modules.a76.invoices.common.common_validators import invoice_exists_by_id
from api.v1.modules.a76.invoices.common.common_validators import invoice_id_required
from api.v1.modules.a76.items.imports.validators.calculations import apply_calculations
from core.exceptions import ErrorCollector
from sqlalchemy import func
@@ -36,7 +36,7 @@ def validate_common(
errors: ErrorCollector,
line_number: int,
):
invoice: InvoiceHeader = invoice_exists_by_id(
invoice: InvoiceHeader = invoice_id_required(
db, line.invoice_id, tenant_id, company_id, errors
)
line_item: LineItem = item_exists(db, line.line_number, tenant_id, company_id)

20
backend/api/v1/modules/a76/items/imports/validators/create.py
View File

@@ -52,8 +52,16 @@ def validate_create(
if not line.class_id:
errors.add_required_error(field=f"line[{line_number}].class_id")
if not line.quantity or not line.quantity.quantity or line.quantity.quantity <= 0:
if not line.quantity or line.quantity.quantity is None:
errors.add_required_error(field=f"line[{line_number}].quantity.quantity")
elif line.quantity.quantity <= 0:
errors.add_error(
field=f"line[{line_number}].quantity.quantity",
message=f"La cantidad debe ser mayor a cero (recibido: {line.quantity.quantity})",
solution=["Capturar una cantidad válida mayor a cero."],
code="INVALID_QUANTITY",
value=float(line.quantity.quantity)
)
# TODO: Añadir validacion SSisGen:CalcularCostoUnitarioEnBaseAValorTotalScaf <-- de la tabla de preferencias de el sistema
# if SSisGen:CalcularCostoUnitarioEnBaseAValorTotalScaf == False:
@@ -67,8 +75,16 @@ def validate_create(
field=f"line[{line_number}].financial.unit_cost_capture"
)
if not line.quantity or not line.quantity.net_weight or line.quantity.net_weight <= 0:
if not line.quantity or line.quantity.net_weight is None:
errors.add_required_error(field=f"line[{line_number}].quantity.net_weight")
elif line.quantity.net_weight <= 0:
errors.add_error(
field=f"line[{line_number}].quantity.net_weight",
message=f"El peso neto debe ser mayor a cero (recibido: {line.quantity.net_weight})",
solution=["Capturar un peso neto válido mayor a cero."],
code="INVALID_NET_WEIGHT",
value=float(line.quantity.net_weight)
)
if not line.customs or not line.customs.origin_country:
errors.add_required_error(field=f"line[{line_number}].customs.origin_country")

36
backend/api/v1/modules/a76/items/imports/validators/update.py
View File

@@ -183,35 +183,51 @@ def validate_update(
line.order = existing_line.order
# Descripciones
if not line.description.description_spanish:
if line.description.description_spanish is None:
line.description.description_spanish = (
existing_line.description.description_spanish
)
if not line.description.description_english:
if line.description.description_english is None:
line.description.description_english = (
existing_line.description.description_english
)
if not line.description.extra_description:
if line.description.extra_description is None:
line.description.extra_description = (
existing_line.description.extra_description
)
# Marca y modelo
if line.description.brand:
# Brand and model
if line.description.brand is not None:
line.description.brand = line.description.brand.upper().strip()
else:
line.description.brand = existing_line.description.brand
if line.description.model:
if line.description.model is not None:
line.description.model = line.description.model.upper().strip()
else:
line.description.model = existing_line.description.model
# Subpartidas (si aplica)
# TODO: Implementar lógica de subpartidas si Loc:LevantarSubpartidas = 'S'
# --- Resolve Settings for inherited parameters ---
from api.v1.modules.a76.app_settings.service import AppSettingsService
settings = AppSettingsService.get_resolved_settings(db, tenant_id, company_id)
inv_type = (invoice.invoice_type or "").strip().upper()
op_type = (invoice.operation_type or "").strip().lower() # 'imp' or 'exp'
# Helper to get nested value from invoices.types.{op}.{type}.ssisgen.ssimpFormData
inv_map = settings.get("invoices", {}).get("types", {}).get(op_type, {}).get(inv_type, {})
# Prefeir ssisgen for this type, then qsisgen, then root ssimpo
form_data = inv_map.get("ssisgen", {}).get("ssimpFormData", {}) or inv_map.get("qsisgen", {}).get("ssimpFormData", {}) or settings.get("ssimpo", {})
# Subpartidas (si aplica)
# Clarion: LOC:LevantarSubpartidas = S
levantar_sub = bool(form_data.get("levantar_subpartidas") or form_data.get("LevantarSubpartidas") or False)
if levantar_sub:
# TODO: Add specific sub-item validation if needed (e.g. parent_line mandatory if it's a subpartida)
# Currently we just ensure the field is carried over if not provided
pass
# Número de parte
if not line.part_number_id:
@@ -229,7 +245,9 @@ def validate_update(
if not line.valuation_method:
if existing_line.valuation_method:
line.valuation_method = existing_line.valuation_method
# else: TODO: Tomar de SisImp:MetValor (preferencias del sistema)
else:
# Tomar de SisImp/SisDef:MetValor (preferencias del sistema)
line.valuation_method = form_data.get("metvalor") or form_data.get("MetValor")
# Número de entrada
if not line.description.entry_number:

21
backend/api/v1/modules/a76/items/models.py
View File

@@ -14,7 +14,6 @@ from core.database import Base
from api.v1.modules.a76.classes.models import Class
from api.v1.modules.a76.general_catalogs.units_of_measure.models import UnitOfMeasure
from api.v1.modules.a76.general_catalogs.identifiers.models import IdentifierDetail
if TYPE_CHECKING:
from .line_financials.models import LineFinancial
@@ -22,8 +21,6 @@ if TYPE_CHECKING:
from .line_customs.models import LineCustom
from .line_descriptions.models import LineDescription
from .line_references.models import LineReference
from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem
from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.invoices.models import InvoiceHeader
@@ -222,7 +219,7 @@ class LineItem(Base, TenantScopedMixin, TimestampMixin):
uselist=False,
)
identifiers: Mapped[List["IdentifierDetail"]] = relationship(
IdentifierDetail,
"IdentifierDetail",
back_populates="line",
cascade="all, delete-orphan",
)
@@ -231,6 +228,11 @@ class LineItem(Base, TenantScopedMixin, TimestampMixin):
foreign_keys=[part_number_id],
viewonly=True,
)
component_part_info: Mapped[Optional["Part"]] = relationship(
"Part",
foreign_keys=[component_part_number_id],
viewonly=True,
)
# ============================================================================
@@ -387,3 +389,14 @@ MIGRATION STRATEGY FROM ORIGINAL TABLES TO NORMALIZED SCHEMA:
)
```
"""
# ============================================================================
# RUNTIME IMPORTS FOR MAPPER RESOLUTION
# ============================================================================
# We import these specialized models at the bottom to ensure they are registered
# in the SQLAlchemy metadata for relationship resolution while avoiding
# circular import issues in the module head.
from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem
from api.v1.modules.a76.parts.models import Part
from api.v1.modules.a76.items.series.models import Serie
from api.v1.modules.a76.general_catalogs.identifiers.models import IdentifierDetail

2
backend/api/v1/modules/a76/items/routes.py
View File

@@ -44,7 +44,7 @@ async def create_item(
- Each LineItem has one LineReference
"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
service = ItemService()
return service.create(db, item_data, tenant_id, company_id)

70
backend/api/v1/modules/a76/items/schemas.py
View File

@@ -290,13 +290,15 @@ class LineItemResponse(LineItemBase):
# Part identification
part_number_id: Optional[int] = Field(
None, alias="part_number", serialization_alias="part_number_id"
None, alias="part_number_id_input", serialization_alias="part_number_id"
)
part_number: Optional[str] = None
component_part_number_id: Optional[int] = Field(
None,
alias="component_part_number",
alias="component_part_number_id_input",
serialization_alias="component_part_number_id",
)
component_part_number: Optional[str] = None
class_id: Optional[int] = None
# Nested data
@@ -325,33 +327,65 @@ class LineItemResponse(LineItemBase):
@model_validator(mode="before")
@classmethod
def extract_relationship_info(cls, data: Any) -> Any:
"""Extract class_code, class_description and unit_of_measure_code from relationships"""
"""Extract information from joined relationships to provide flat mapping for UI."""
if isinstance(data, dict):
# If already a dict, ensure description syncs to top-level if missing
desc = data.get("description", {})
if isinstance(desc, dict):
if not data.get("part_description_es"):
data["part_description_es"] = desc.get("description_spanish")
if not data.get("part_description_en"):
data["part_description_en"] = desc.get("description_english")
return data
# It's an ORM object
result = {}
for key in cls.model_fields.keys():
if hasattr(data, key):
result[key] = getattr(data, key)
# 1. Start with model attributes (columns)
if hasattr(data, "__table__"):
for k in data.__table__.columns.keys():
result[k] = getattr(data, k, None)
else:
# Fallback for non-table objects if any
for k, v in data.__dict__.items():
if not k.startswith("_"):
result[k] = v
# Map model field names to schema field names for aliased fields
if hasattr(data, "part_number"):
result["part_number_id"] = data.part_number
if hasattr(data, "component_part_number"):
result["component_part_number_id"] = data.component_part_number
# Alias mapping for part numbers
if hasattr(data, "part_number_id") and "part_number_id" not in result:
result["part_number_id"] = data.part_number_id
if hasattr(data, "component_part_number_id") and "component_part_number_id" not in result:
result["component_part_number_id"] = data.component_part_number_id
# Extract part info (string part numbers) from relationship objects
if hasattr(data, "part_info") and data.part_info is not None:
result["part_number"] = getattr(data.part_info, "part_number", None)
if hasattr(data, "component_part_info") and data.component_part_info is not None:
result["component_part_number"] = getattr(data.component_part_info, "part_number", None)
# Extract class info
if hasattr(data, "class_info") and data.class_info is not None:
result["class_code"] = data.class_info.class_code
result["class_description"] = data.class_info.description_es
result["class_code"] = getattr(data.class_info, "class_code", None)
result["class_description"] = getattr(data.class_info, "description_es", None)
# Extract unit of measure code
if (
hasattr(data, "unit_of_measure_info")
and data.unit_of_measure_info is not None
):
result["unit_of_measure_code"] = data.unit_of_measure_info.code
if hasattr(data, "unit_of_measure_info") and data.unit_of_measure_info is not None:
result["unit_of_measure_code"] = getattr(data.unit_of_measure_info, "code", None)
# 2. Extract nested objects and populate redundant descriptions
# We MUST use explicit getattr for relationships to ensure SQLAlchemy loads/uses joined-loaded ones
for key in ["financial", "quantity", "customs", "description", "reference", "fa_data", "series", "identifiers"]:
val = getattr(data, key, None)
if val is not None:
result[key] = val
# Sync to top-level for description redundancy (huge boost for UI stability)
if key == "description":
result["part_description_es"] = getattr(val, "description_spanish", None)
result["part_description_en"] = getattr(val, "description_english", None)
else:
# Provide default empty dict for core containers to help frontend
if key in ["financial", "quantity", "customs", "description"]:
result[key] = {}
return result

5
backend/api/v1/modules/a76/items/series/models.py
View File

@@ -16,10 +16,13 @@ class Serie(Base, TenantScopedMixin, TimestampMixin):
serial_numbers: Mapped[Optional[str]] = mapped_column(String(50)) # SERIEEXPO
model: Mapped[Optional[str]] = mapped_column(String(50)) # MODELOEXPO
sub_model: Mapped[Optional[str]] = mapped_column(String(50)) # SUBMODELOEXPO
brand: Mapped[Optional[str]] = mapped_column(String(50)) # MARCAEXPO
brand: Mapped[Optional[str]] = mapped_column(String(50)) # MARCA
# expo_brand: Mapped[Optional[str]] = mapped_column(String(50)) # MARCAEXPO
number_id: Mapped[Optional[str]] = mapped_column(String(25)) # NUMIDEXPO
discharge: Mapped[Optional[bool]] = mapped_column(Boolean) # MARCA
serie_row: Mapped[Optional[int]] = mapped_column(Integer) # LINEASERIEIMPO <-- IN CASE OF EXPO
# import_invoice: Mapped[Optional[str]] = mapped_column(String(15)) # FACTURAIMPO
# import_line: Mapped[Optional[int]] = mapped_column(Integer) # LINEAIMPO
image_path: Mapped[Optional[str]] = mapped_column(String(255)) # PATH DE IMAGEN (MEX)

6
backend/api/v1/modules/a76/items/series/schemas.py
View File

@@ -8,10 +8,10 @@ class SerieBase(BaseModel):
model: Optional[str] = Field(None, max_length=50, description="Model (MODELOEXPO)")
sub_model: Optional[str] = Field(None, max_length=50, description="Sub model (SUBMODELOEXPO)")
brand: Optional[str] = Field(None, max_length=50, description="Brand (MARCA)")
expo_brad: Optional[str] = Field(None, max_length=50, description="Expo brand (MARCAEXPO)")
# expo_brand: Optional[str] = Field(None, max_length=50, description="Expo brand (MARCAEXPO)")
number_id: Optional[str] = Field(None, max_length=25, description="Number ID (NUMIDEXPO)")
import_invoice: Optional[str] = Field(None, max_length=15, description="Import invoice (FACTURAIMPO)")
import_line: Optional[int] = Field(None, description="Import line (LINEAIMPO)")
# import_invoice: Optional[str] = Field(None, max_length=15, description="Import invoice (FACTURAIMPO)")
# import_line: Optional[int] = Field(None, description="Import line (LINEAIMPO)")
image_path: Optional[str] = Field(None, max_length=255, description="Image path (IMAGEPATHMEX)")

198
backend/api/v1/modules/a76/items/service.py
View File

@@ -15,7 +15,7 @@ There is no intermediate Item entity anymore. Each LineItem belongs directly to
import datetime
import logging
from decimal import Decimal
from typing import Optional, List, Tuple
from typing import Any, Optional, List, Tuple
from fastapi import HTTPException
from sqlalchemy import and_, case, func, or_, select
from sqlalchemy.exc import IntegrityError
@@ -145,6 +145,16 @@ class ItemService:
)
return None
@staticmethod
def _filter_model_data(data: dict, model_class: Any) -> dict:
"""Filter a dictionary to only include keys that exist as attributes in the model class."""
if not data:
return {}
from sqlalchemy import inspect
mapper = inspect(model_class)
valid_keys = set(mapper.columns.keys())
return {k: v for k, v in data.items() if k in valid_keys}
@staticmethod
def _create_line_nested_data(
db: Session, line: LineItem, line_data, tenant_id: int, company_id: int
@@ -166,7 +176,9 @@ class ItemService:
else data.model_dump()
)
nested_dict["item_line_id"] = line.id
db.add(model_class(**nested_dict))
# Filter dict against model attributes
filtered_dict = ItemService._filter_model_data(nested_dict, model_class)
db.add(model_class(**filtered_dict))
# FA data uses line.id as primary key
if line_data.fa_data:
@@ -185,7 +197,8 @@ class ItemService:
if isinstance(line_data.series, list)
else [line_data.series]
)
for s in series_list:
new_series = []
for i, s in enumerate(series_list):
serie_dict = (
s.model_dump(exclude_unset=True)
if hasattr(s, "model_dump")
@@ -193,12 +206,16 @@ class ItemService:
)
if not serie_dict:
continue
serie_dict["line_item_id"] = line.id
serie_dict["tenant_id"] = tenant_id
serie_dict["company_id"] = company_id
serie_dict.update({
"tenant_id": tenant_id,
"company_id": company_id
})
if serie_dict.get("row") is None:
serie_dict["row"] = 1
db.add(Serie(**serie_dict))
serie_dict["row"] = i + 1
# Filter dict against model attributes
filtered_s = ItemService._filter_model_data(serie_dict, Serie)
db.add(Serie(**filtered_s))
# Identifier Detail data
if hasattr(line_data, "identifiers") and line_data.identifiers:
@@ -207,6 +224,7 @@ class ItemService:
if isinstance(line_data.identifiers, list)
else [line_data.identifiers]
)
new_ids = []
for d in id_list:
id_dict = (
d.model_dump(exclude_unset=True)
@@ -215,10 +233,15 @@ class ItemService:
)
if not id_dict:
continue
id_dict["item_line_id"] = line.id
id_dict["tenant_id"] = tenant_id
id_dict["company_id"] = company_id
db.add(IdentifierDetail(**id_dict))
id_dict.update({
"tenant_id": tenant_id,
"company_id": company_id
})
# Filter dict against model attributes
filtered_id = ItemService._filter_model_data(id_dict, IdentifierDetail)
new_ids.append(IdentifierDetail(**filtered_id))
line.identifiers = new_ids
@staticmethod
def _attach_series(db: Session, item: LineItem) -> None:
@@ -241,6 +264,7 @@ class ItemService:
)
item.identifiers = list(identifiers)
@staticmethod
def get_by_id(
db: Session, item_id: int, tenant_id: int, company_id: int
@@ -257,6 +281,8 @@ class ItemService:
joinedload(LineItem.class_info),
joinedload(LineItem.unit_of_measure_info),
joinedload(LineItem.fa_data),
joinedload(LineItem.part_info),
joinedload(LineItem.component_part_info),
)
.filter(
LineItem.id == item_id,
@@ -293,6 +319,8 @@ class ItemService:
joinedload(LineItem.class_info),
joinedload(LineItem.unit_of_measure_info),
joinedload(LineItem.fa_data),
joinedload(LineItem.part_info),
joinedload(LineItem.component_part_info),
)
.filter(
LineItem.tenant_id == tenant_id,
@@ -366,6 +394,8 @@ class ItemService:
joinedload(LineItem.description),
joinedload(LineItem.reference),
joinedload(LineItem.fa_data),
joinedload(LineItem.part_info),
joinedload(LineItem.component_part_info),
)
.filter(
LineItem.invoice_id == invoice_id,
@@ -505,20 +535,42 @@ class ItemService:
)
# Create the item
# Filter main item_dict against LineItem model attributes
item_dict = ItemService._filter_model_data(item_dict, LineItem)
db_item = LineItem(**item_dict)
db.add(db_item)
db.flush() # Get the item ID
# Create all nested data
# Create nested data
ItemService._create_line_nested_data(
db, db_item, item_data, tenant_id, company_id
)
db.commit()
db.refresh(db_item)
ItemService._attach_series(db, db_item)
ItemService._attach_identifiers(db, db_item)
return db_item
# Eager load EVERYTHING needed for the response before returning
final_item = (
db.query(LineItem)
.options(
joinedload(LineItem.financial),
joinedload(LineItem.quantity),
joinedload(LineItem.customs),
joinedload(LineItem.description),
joinedload(LineItem.reference),
joinedload(LineItem.class_info),
joinedload(LineItem.unit_of_measure_info),
joinedload(LineItem.fa_data),
joinedload(LineItem.part_info),
joinedload(LineItem.component_part_info),
)
.filter(LineItem.id == db_item.id)
.first()
)
if final_item:
ItemService._attach_series(db, final_item)
ItemService._attach_identifiers(db, final_item)
return final_item
except IntegrityError as e:
db.rollback()
@@ -648,53 +700,111 @@ class ItemService:
"reference",
"fa_data",
"series",
"identifiers",
},
exclude_unset=True,
)
# Update item fields
# CONDITIONAL update of nested data to prevent data loss
# Perform in-place updates for one-to-one relations, full replacement for one-to-many
# Update item attributes
# Filter main item_dict against LineItem model attributes
item_dict = ItemService._filter_model_data(item_dict, LineItem)
for key, value in item_dict.items():
setattr(db_item, key, value)
# Delete existing nested data
db.query(LineFinancial).filter(
LineFinancial.item_line_id == db_item.id
).delete()
db.query(LineQuantity).filter(
LineQuantity.item_line_id == db_item.id
).delete()
db.query(LineCustom).filter(LineCustom.item_line_id == db_item.id).delete()
db.query(LineDescription).filter(
LineDescription.item_line_id == db_item.id
).delete()
db.query(LineReference).filter(
LineReference.item_line_id == db_item.id
).delete()
db.query(FaLineItem).filter(FaLineItem.id == db_item.id).delete()
db.query(Serie).filter(Serie.line_item_id == db_item.id).delete()
db.query(IdentifierDetail).filter(IdentifierDetail.item_line_id == db_item.id).delete()
db.flush()
# 2. Update nested one-to-one objects (In-place update)
nested_relations = [
('financial', LineFinancial, 'item_line_id'),
('quantity', LineQuantity, 'item_line_id'),
('customs', LineCustom, 'item_line_id'),
('description', LineDescription, 'item_line_id'),
('reference', LineReference, 'item_line_id')
]
# Create new nested data
ItemService._create_line_nested_data(
db, db_item, item_data, tenant_id, company_id
)
for attr_name, model_class, fk_name in nested_relations:
attr_data = getattr(item_data, attr_name)
if attr_data is not None:
db_nested = getattr(db_item, attr_name)
nested_dict = attr_data.model_dump(exclude_unset=True)
if db_nested:
# Update existing
for k, v in nested_dict.items():
setattr(db_nested, k, v)
else:
# Create new
nested_dict[fk_name] = db_item.id
new_nested = model_class(**nested_dict)
setattr(db_item, attr_name, new_nested)
db.add(new_nested)
# 3. Handle fa_data (special case as PK is shared)
if item_data.fa_data is not None:
fa_dict = item_data.fa_data.model_dump(
exclude_unset=True, exclude={"line_item_id", "includes_subitems"}
)
if db_item.fa_data:
for k, v in fa_dict.items():
setattr(db_item.fa_data, k, v)
else:
fa_dict.update({
"id": db_item.id,
"tenant_id": tenant_id,
"company_id": company_id
})
db_item.fa_data = FaLineItem(**fa_dict)
db.add(db_item.fa_data)
# 4. Handle one-to-many arrays (Full replacement as these are collections)
if item_data.series is not None:
# Use synchronize_session='fetch' to ensure the session knows about the deletions
db.query(Serie).filter(Serie.line_item_id == db_item.id).delete(synchronize_session='fetch')
for s_data in item_data.series:
s_dict = s_data.model_dump(exclude_unset=True)
s_dict.update({
"line_item_id": db_item.id,
"tenant_id": tenant_id,
"company_id": company_id
})
if s_dict.get("row") is None:
s_dict["row"] = 1
# Filter dict against model attributes
filtered_s = ItemService._filter_model_data(s_dict, Serie)
db.add(Serie(**filtered_s))
if item_data.identifiers is not None:
db.query(IdentifierDetail).filter(IdentifierDetail.item_line_id == db_item.id).delete()
for d in item_data.identifiers:
id_dict = d.model_dump(exclude_unset=True)
id_dict.update({
"item_line_id": db_item.id,
"tenant_id": tenant_id,
"company_id": company_id
})
# Filter dict against model attributes
filtered_id = ItemService._filter_model_data(id_dict, IdentifierDetail)
db.add(IdentifierDetail(**filtered_id))
db.flush()
# Renumber all lines for this invoice to ensure consecutive numbering
ItemService._renumber_all_invoice_lines(db, db_item.invoice_id)
db.commit()
db.refresh(db_item)
ItemService._attach_series(db, db_item)
ItemService._attach_identifiers(db, db_item)
db.refresh(db_item, ["financial", "quantity", "customs", "description", "reference", "fa_data", "identifiers"])
return db_item
except HTTPException:
raise
except Exception as e:
db.rollback()
import traceback
logger.error(f"Unexpected error updating item: {e}")
raise HTTPException(status_code=500, detail="Error updating item")
raise HTTPException(status_code=500, detail=f"Error updating item: {str(e)}")
@staticmethod
def delete(