Merge pull request 'feat: Implement CRUD operations for FDA catalog entries and their associated specifications, constituent elements, affirmation codes, and lot productions.' (#176) from feature/FDA into development

Reviewed-on: ADUANASOFT/anexo76#176
This commit is contained in:
2026-03-03 14:19:42 +00:00
16 changed files with 2587 additions and 9 deletions

View File

@@ -39,6 +39,7 @@ class FDACatalogUpdate(BaseModel):
}
class FDACatalogResponse(BaseModel):
"""DTO para respuesta del catálogo FDA"""
id: int
@@ -51,9 +52,121 @@ class FDACatalogResponse(BaseModel):
storage_status: Optional[str]
warehouse_code: Optional[str]
call_atl: Optional[str]
created_at: Optional[str]
updated_at: Optional[str]
model_config = {
"from_attributes": True
}
# DTOs para FDA01 - Especificaciones
class FDASpecificationsBase(BaseModel):
prod_code: Optional[str] = None
commodity_desc: Optional[str] = None
brand_name: Optional[str] = None
disclaimer: Optional[str] = None
pgm_code: Optional[str] = None
proc_code: Optional[str] = None
intnd_use_code: Optional[str] = None
intnd_use_desc: Optional[str] = None
temp_qual: Optional[str] = None
temp_type: Optional[str] = None
temp_degrees: Optional[float] = None
temp_negative: Optional[float] = None
temp_location: Optional[str] = None
quantity_1: Optional[float] = None
qty_uom_1: Optional[str] = None
quantity_2: Optional[float] = None
qty_uom_2: Optional[str] = None
quantity_3: Optional[float] = None
qty_uom_3: Optional[str] = None
ctry_prod: Optional[str] = None
ctry_source: Optional[str] = None
ctry_growth: Optional[str] = None
ctry_refusal: Optional[str] = None
ctry_shipping: Optional[str] = None
manuf_key: Optional[str] = None
shipper_key: Optional[str] = None
ult_cons_key: Optional[str] = None
fda_imp_key: Optional[str] = None
pn_subm_key: Optional[str] = None
consol_key: Optional[str] = None
producer_key: Optional[str] = None
owner_key: Optional[str] = None
deli_party_key: Optional[str] = None
grower_key: Optional[str] = None
dev_ini_imp_key: Optional[str] = None
lacf_cont_1: Optional[str] = None
lacf_cont_2: Optional[str] = None
lacf_cont_3: Optional[str] = None
pn_transmitter_key: Optional[str] = None
class FDASpecificationsCreate(FDASpecificationsBase):
pass
class FDASpecificationsUpdate(FDASpecificationsBase):
pass
class FDASpecificationsResponse(FDASpecificationsBase):
id: int
fda_catalog_id: int
model_config = {"from_attributes": True}
# DTOs para FDA04 - Elementos Constitutivos
class FDAConstituentElementsBase(BaseModel):
line: int
ele_name: str
ele_qty: Optional[float] = None
ele_qty_uom: Optional[str] = None
ele_pctg: Optional[float] = None
class FDAConstituentElementsCreate(FDAConstituentElementsBase):
pass
class FDAConstituentElementsUpdate(FDAConstituentElementsBase):
line: Optional[int] = None
ele_name: Optional[str] = None
class FDAConstituentElementsResponse(FDAConstituentElementsBase):
id: int
fda_catalog_id: int
model_config = {"from_attributes": True}
# DTOs para FDA23 - Códigos de Afirmación
class FDAAffirmationCodesBase(BaseModel):
line: int
aoc_code: str
aoc_qual: Optional[str] = None
class FDAAffirmationCodesCreate(FDAAffirmationCodesBase):
pass
class FDAAffirmationCodesUpdate(FDAAffirmationCodesBase):
line: Optional[int] = None
aoc_code: Optional[str] = None
class FDAAffirmationCodesResponse(FDAAffirmationCodesBase):
id: int
fda_catalog_id: int
model_config = {"from_attributes": True}
# DTOs para FDA25 - Lotes de Producción
class FDALotProductionBase(BaseModel):
line: int
lot_number: str
production_start_date: Optional[str] = None
production_end_date: Optional[str] = None
class FDALotProductionCreate(FDALotProductionBase):
pass
class FDALotProductionUpdate(FDALotProductionBase):
line: Optional[int] = None
lot_number: Optional[str] = None
class FDALotProductionResponse(FDALotProductionBase):
id: int
fda_catalog_id: int
model_config = {"from_attributes": True}

View File

@@ -1,9 +1,9 @@
"""
Modelo para el catálogo de claves FDA
"""
from typing import Optional
from sqlalchemy import String, Integer, UniqueConstraint, PrimaryKeyConstraint
from sqlalchemy.orm import Mapped, mapped_column
from typing import Optional, List
from sqlalchemy import String, Integer, Numeric, Date, ForeignKey, UniqueConstraint, PrimaryKeyConstraint
from sqlalchemy.orm import Mapped, mapped_column, relationship
from core.database import Base
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
@@ -27,3 +27,124 @@ class FDACatalog(Base, TenantScopedMixin, TimestampMixin):
storage_status: Mapped[Optional[str]] = mapped_column(String(100))
warehouse_code: Mapped[Optional[str]] = mapped_column(String(20))
call_atl: Mapped[Optional[str]] = mapped_column(String(20))
# Relationships
specifications: Mapped[Optional["FDASpecifications"]] = relationship("FDASpecifications", back_populates="fda_catalog", cascade="all, delete-orphan", uselist=False)
constituent_elements: Mapped[List["FDAConstituentElements"]] = relationship("FDAConstituentElements", back_populates="fda_catalog", cascade="all, delete-orphan")
affirmation_codes: Mapped[List["FDAAffirmationCodes"]] = relationship("FDAAffirmationCodes", back_populates="fda_catalog", cascade="all, delete-orphan")
lot_productions: Mapped[List["FDALotProduction"]] = relationship("FDALotProduction", back_populates="fda_catalog", cascade="all, delete-orphan")
class FDASpecifications(Base, TenantScopedMixin, TimestampMixin):
"""FDA01 - Especificaciones del Producto FDA"""
__tablename__ = "fda_specifications"
__table_args__ = (
PrimaryKeyConstraint("id", name="fda_specifications_pkey"),
{'schema': 'a76'}
)
id: Mapped[int] = mapped_column(Integer, primary_key=True)
fda_catalog_id: Mapped[int] = mapped_column(ForeignKey("a76.fda_catalog.id", ondelete="CASCADE"), nullable=False, index=True)
prod_code: Mapped[Optional[str]] = mapped_column(String(50))
commodity_desc: Mapped[Optional[str]] = mapped_column(String(200))
brand_name: Mapped[Optional[str]] = mapped_column(String(100))
disclaimer: Mapped[Optional[str]] = mapped_column(String(100))
pgm_code: Mapped[Optional[str]] = mapped_column(String(50))
proc_code: Mapped[Optional[str]] = mapped_column(String(50))
intnd_use_code: Mapped[Optional[str]] = mapped_column(String(50))
intnd_use_desc: Mapped[Optional[str]] = mapped_column(String(200))
temp_qual: Mapped[Optional[str]] = mapped_column(String(50))
temp_type: Mapped[Optional[str]] = mapped_column(String(50))
temp_degrees: Mapped[Optional[float]] = mapped_column(Numeric(10, 2))
temp_negative: Mapped[Optional[float]] = mapped_column(Numeric(10, 2))
temp_location: Mapped[Optional[str]] = mapped_column(String(100))
quantity_1: Mapped[Optional[float]] = mapped_column(Numeric(15, 2))
qty_uom_1: Mapped[Optional[str]] = mapped_column(String(20))
quantity_2: Mapped[Optional[float]] = mapped_column(Numeric(15, 2))
qty_uom_2: Mapped[Optional[str]] = mapped_column(String(20))
quantity_3: Mapped[Optional[float]] = mapped_column(Numeric(15, 2))
qty_uom_3: Mapped[Optional[str]] = mapped_column(String(20))
ctry_prod: Mapped[Optional[str]] = mapped_column(String(50))
ctry_source: Mapped[Optional[str]] = mapped_column(String(50))
ctry_growth: Mapped[Optional[str]] = mapped_column(String(50))
ctry_refusal: Mapped[Optional[str]] = mapped_column(String(50))
ctry_shipping: Mapped[Optional[str]] = mapped_column(String(50))
manuf_key: Mapped[Optional[str]] = mapped_column(String(50))
shipper_key: Mapped[Optional[str]] = mapped_column(String(50))
ult_cons_key: Mapped[Optional[str]] = mapped_column(String(50))
fda_imp_key: Mapped[Optional[str]] = mapped_column(String(50))
pn_subm_key: Mapped[Optional[str]] = mapped_column(String(50))
consol_key: Mapped[Optional[str]] = mapped_column(String(50))
producer_key: Mapped[Optional[str]] = mapped_column(String(50))
owner_key: Mapped[Optional[str]] = mapped_column(String(50))
deli_party_key: Mapped[Optional[str]] = mapped_column(String(50))
grower_key: Mapped[Optional[str]] = mapped_column(String(50))
dev_ini_imp_key: Mapped[Optional[str]] = mapped_column(String(50))
lacf_cont_1: Mapped[Optional[str]] = mapped_column(String(100))
lacf_cont_2: Mapped[Optional[str]] = mapped_column(String(100))
lacf_cont_3: Mapped[Optional[str]] = mapped_column(String(100))
pn_transmitter_key: Mapped[Optional[str]] = mapped_column(String(50))
fda_catalog: Mapped["FDACatalog"] = relationship("FDACatalog", back_populates="specifications")
class FDAConstituentElements(Base, TenantScopedMixin, TimestampMixin):
"""FDA04 - Constituent Elements"""
__tablename__ = "fda_constituent_elements"
__table_args__ = (
PrimaryKeyConstraint("id", name="fda_constituent_elements_pkey"),
{'schema': 'a76'}
)
id: Mapped[int] = mapped_column(Integer, primary_key=True)
fda_catalog_id: Mapped[int] = mapped_column(ForeignKey("a76.fda_catalog.id", ondelete="CASCADE"), nullable=False, index=True)
line: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
ele_name: Mapped[str] = mapped_column(String(200), nullable=False)
ele_qty: Mapped[Optional[float]] = mapped_column(Numeric(15, 2))
ele_qty_uom: Mapped[Optional[str]] = mapped_column(String(20))
ele_pctg: Mapped[Optional[float]] = mapped_column(Numeric(15, 2))
fda_catalog: Mapped["FDACatalog"] = relationship("FDACatalog", back_populates="constituent_elements")
class FDAAffirmationCodes(Base, TenantScopedMixin, TimestampMixin):
"""FDA23 - Affirmation of Compliance"""
__tablename__ = "fda_affirmation_codes"
__table_args__ = (
PrimaryKeyConstraint("id", name="fda_affirmation_codes_pkey"),
{'schema': 'a76'}
)
id: Mapped[int] = mapped_column(Integer, primary_key=True)
fda_catalog_id: Mapped[int] = mapped_column(ForeignKey("a76.fda_catalog.id", ondelete="CASCADE"), nullable=False, index=True)
line: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
aoc_code: Mapped[str] = mapped_column(String(50), nullable=False)
aoc_qual: Mapped[Optional[str]] = mapped_column(String(100))
fda_catalog: Mapped["FDACatalog"] = relationship("FDACatalog", back_populates="affirmation_codes")
class FDALotProduction(Base, TenantScopedMixin, TimestampMixin):
"""FDA25 - Lot and Production Dates"""
__tablename__ = "fda_lot_production"
__table_args__ = (
PrimaryKeyConstraint("id", name="fda_lot_production_pkey"),
{'schema': 'a76'}
)
id: Mapped[int] = mapped_column(Integer, primary_key=True)
fda_catalog_id: Mapped[int] = mapped_column(ForeignKey("a76.fda_catalog.id", ondelete="CASCADE"), nullable=False, index=True)
line: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
lot_number: Mapped[str] = mapped_column(String(100), nullable=False)
production_start_date: Mapped[Optional[str]] = mapped_column(String(50))
production_end_date: Mapped[Optional[str]] = mapped_column(String(50))
fda_catalog: Mapped["FDACatalog"] = relationship("FDACatalog", back_populates="lot_productions")

View File

@@ -67,3 +67,204 @@ async def get_fda_catalog(
"warehouse_code": entry.warehouse_code,
"call_atl": entry.call_atl
}
from api.v1.modules.a76.general_catalogs.fda_catalog.dto import FDACatalogCreate, FDACatalogUpdate
from sqlalchemy.exc import IntegrityError
@router.post("/", response_model=FDACatalogResponse, status_code=status.HTTP_201_CREATED)
async def create_fda_catalog(
data: FDACatalogCreate,
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user)
):
"""Crear nueva entrada en el catálogo FDA"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
try:
return FDACatalogService.create(db, tenant_id, company_id, data)
except IntegrityError:
db.rollback()
raise HTTPException(status_code=400, detail="La Clave FDA ya existe o hay un conflicto de datos.")
@router.put("/{id}", response_model=FDACatalogResponse)
async def update_fda_catalog(
id: int,
data: FDACatalogUpdate,
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user)
):
"""Actualizar entrada del catálogo FDA"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
try:
entry = FDACatalogService.update(db, tenant_id, company_id, id, data)
if not entry:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Entrada no encontrada")
return entry
except IntegrityError:
db.rollback()
raise HTTPException(status_code=400, detail="Error de integridad de datos al actualizar la Clave FDA.")
@router.delete("/{id}")
async def delete_fda_catalog(
id: int,
company_id: int = Query(..., description="Company ID"),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user)
):
"""Eliminar entrada del catálogo FDA"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
success = FDACatalogService.delete(db, tenant_id, company_id, id)
if not success:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Entrada no encontrada")
return {"message": "Eliminado correctamente"}
from typing import List
from api.v1.modules.a76.general_catalogs.fda_catalog.service import FDADetailsService
from api.v1.modules.a76.general_catalogs.fda_catalog.dto import (
FDASpecificationsCreate, FDASpecificationsResponse,
FDAConstituentElementsCreate, FDAConstituentElementsUpdate, FDAConstituentElementsResponse,
FDAAffirmationCodesCreate, FDAAffirmationCodesUpdate, FDAAffirmationCodesResponse,
FDALotProductionCreate, FDALotProductionUpdate, FDALotProductionResponse
)
# === FDA01: Especificaciones del Producto ===
@router.get("/{id}/specifications", response_model=Optional[FDASpecificationsResponse])
async def get_fda_specifications(
id: int,
company_id: int = Query(..., description="Company ID"),
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)
catalog_entry = FDACatalogService.get_by_id(db, tenant_id, company_id, id)
if not catalog_entry:
raise HTTPException(status_code=404, detail="Catálogo FDA no encontrado")
return FDADetailsService.get_specifications(db, tenant_id, company_id, id)
@router.put("/{id}/specifications", response_model=FDASpecificationsResponse)
async def save_fda_specifications(
id: int,
data: FDASpecificationsCreate,
company_id: int = Query(..., description="Company ID"),
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)
catalog_entry = FDACatalogService.get_by_id(db, tenant_id, company_id, id)
if not catalog_entry:
raise HTTPException(status_code=404, detail="Catálogo FDA no encontrado")
return FDADetailsService.save_specifications(db, tenant_id, company_id, id, data.model_dump(exclude_unset=True))
# === FDA04: Elementos Constitutivos ===
@router.get("/{id}/constituent-elements", response_model=List[FDAConstituentElementsResponse])
async def list_fda_constituent_elements(
id: int,
company_id: int = Query(..., description="Company ID"),
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)
return FDADetailsService.get_constituent_elements(db, tenant_id, company_id, id)
@router.post("/{id}/constituent-elements", response_model=FDAConstituentElementsResponse)
async def create_fda_constituent_element(
id: int,
data: FDAConstituentElementsCreate,
company_id: int = Query(..., description="Company ID"),
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)
return FDADetailsService.create_constituent_element(db, tenant_id, company_id, id, data.model_dump())
@router.delete("/{id}/constituent-elements/{element_id}")
async def delete_fda_constituent_element(
id: int,
element_id: int,
company_id: int = Query(..., description="Company ID"),
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)
success = FDADetailsService.delete_constituent_element(db, tenant_id, company_id, element_id)
if not success:
raise HTTPException(status_code=404, detail="Elemento no encontrado")
return {"message": "Eliminado correctamente"}
# === FDA23: Códigos de Afirmación ===
@router.get("/{id}/affirmation-codes", response_model=List[FDAAffirmationCodesResponse])
async def list_fda_affirmation_codes(
id: int,
company_id: int = Query(..., description="Company ID"),
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)
return FDADetailsService.get_affirmation_codes(db, tenant_id, company_id, id)
@router.post("/{id}/affirmation-codes", response_model=FDAAffirmationCodesResponse)
async def create_fda_affirmation_code(
id: int,
data: FDAAffirmationCodesCreate,
company_id: int = Query(..., description="Company ID"),
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)
return FDADetailsService.create_affirmation_code(db, tenant_id, company_id, id, data.model_dump())
@router.delete("/{id}/affirmation-codes/{code_id}")
async def delete_fda_affirmation_code(
id: int,
code_id: int,
company_id: int = Query(..., description="Company ID"),
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)
success = FDADetailsService.delete_affirmation_code(db, tenant_id, company_id, code_id)
if not success:
raise HTTPException(status_code=404, detail="Código no encontrado")
return {"message": "Eliminado correctamente"}
# === FDA25: Lotes de Producción ===
@router.get("/{id}/lot-productions", response_model=List[FDALotProductionResponse])
async def list_fda_lot_productions(
id: int,
company_id: int = Query(..., description="Company ID"),
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)
return FDADetailsService.get_lot_productions(db, tenant_id, company_id, id)
@router.post("/{id}/lot-productions", response_model=FDALotProductionResponse)
async def create_fda_lot_production(
id: int,
data: FDALotProductionCreate,
company_id: int = Query(..., description="Company ID"),
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)
return FDADetailsService.create_lot_production(db, tenant_id, company_id, id, data.model_dump())
@router.delete("/{id}/lot-productions/{lot_id}")
async def delete_fda_lot_production(
id: int,
lot_id: int,
company_id: int = Query(..., description="Company ID"),
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)
success = FDADetailsService.delete_lot_production(db, tenant_id, company_id, lot_id)
if not success:
raise HTTPException(status_code=404, detail="Lote no encontrado")
return {"message": "Eliminado correctamente"}

View File

@@ -106,3 +106,162 @@ class FDACatalogService:
db.delete(entry)
db.commit()
return True
class FDADetailsService:
"""Servicio CRUD para los detalles adicionales de FDA (Especificaciones, Lotes, etc)"""
# FDA01 - Especificaciones
@staticmethod
def get_specifications(db: Session, tenant_id: int, company_id: int, fda_catalog_id: int):
from api.v1.modules.a76.general_catalogs.fda_catalog.models import FDASpecifications
return db.execute(
select(FDASpecifications).where(
(FDASpecifications.fda_catalog_id == fda_catalog_id) &
(FDASpecifications.tenant_id == tenant_id) &
(FDASpecifications.company_id == company_id)
)
).scalar_one_or_none()
@staticmethod
def save_specifications(db: Session, tenant_id: int, company_id: int, fda_catalog_id: int, data: dict):
from api.v1.modules.a76.general_catalogs.fda_catalog.models import FDASpecifications
existing = FDADetailsService.get_specifications(db, tenant_id, company_id, fda_catalog_id)
if existing:
for field, value in data.items():
setattr(existing, field, value)
entry = existing
else:
entry = FDASpecifications(
tenant_id=tenant_id,
company_id=company_id,
fda_catalog_id=fda_catalog_id,
**data
)
db.add(entry)
db.commit()
db.refresh(entry)
return entry
# FDA04 - Elementos Constitutivos
@staticmethod
def get_constituent_elements(db: Session, tenant_id: int, company_id: int, fda_catalog_id: int):
from api.v1.modules.a76.general_catalogs.fda_catalog.models import FDAConstituentElements
items = db.execute(
select(FDAConstituentElements).where(
(FDAConstituentElements.fda_catalog_id == fda_catalog_id) &
(FDAConstituentElements.tenant_id == tenant_id) &
(FDAConstituentElements.company_id == company_id)
).order_by(FDAConstituentElements.line.asc())
).scalars().all()
return items
@staticmethod
def create_constituent_element(db: Session, tenant_id: int, company_id: int, fda_catalog_id: int, data: dict):
from api.v1.modules.a76.general_catalogs.fda_catalog.models import FDAConstituentElements
entry = FDAConstituentElements(
tenant_id=tenant_id,
company_id=company_id,
fda_catalog_id=fda_catalog_id,
**data
)
db.add(entry)
db.commit()
db.refresh(entry)
return entry
@staticmethod
def delete_constituent_element(db: Session, tenant_id: int, company_id: int, element_id: int):
from api.v1.modules.a76.general_catalogs.fda_catalog.models import FDAConstituentElements
entry = db.execute(select(FDAConstituentElements).where(
(FDAConstituentElements.id == element_id) &
(FDAConstituentElements.tenant_id == tenant_id) &
(FDAConstituentElements.company_id == company_id)
)).scalar_one_or_none()
if entry:
db.delete(entry)
db.commit()
return True
return False
# FDA23 - Códigos de Afirmación
@staticmethod
def get_affirmation_codes(db: Session, tenant_id: int, company_id: int, fda_catalog_id: int):
from api.v1.modules.a76.general_catalogs.fda_catalog.models import FDAAffirmationCodes
items = db.execute(
select(FDAAffirmationCodes).where(
(FDAAffirmationCodes.fda_catalog_id == fda_catalog_id) &
(FDAAffirmationCodes.tenant_id == tenant_id) &
(FDAAffirmationCodes.company_id == company_id)
).order_by(FDAAffirmationCodes.line.asc())
).scalars().all()
return items
@staticmethod
def create_affirmation_code(db: Session, tenant_id: int, company_id: int, fda_catalog_id: int, data: dict):
from api.v1.modules.a76.general_catalogs.fda_catalog.models import FDAAffirmationCodes
entry = FDAAffirmationCodes(
tenant_id=tenant_id,
company_id=company_id,
fda_catalog_id=fda_catalog_id,
**data
)
db.add(entry)
db.commit()
db.refresh(entry)
return entry
@staticmethod
def delete_affirmation_code(db: Session, tenant_id: int, company_id: int, code_id: int):
from api.v1.modules.a76.general_catalogs.fda_catalog.models import FDAAffirmationCodes
entry = db.execute(select(FDAAffirmationCodes).where(
(FDAAffirmationCodes.id == code_id) &
(FDAAffirmationCodes.tenant_id == tenant_id) &
(FDAAffirmationCodes.company_id == company_id)
)).scalar_one_or_none()
if entry:
db.delete(entry)
db.commit()
return True
return False
# FDA25 - Lotes de Producción
@staticmethod
def get_lot_productions(db: Session, tenant_id: int, company_id: int, fda_catalog_id: int):
from api.v1.modules.a76.general_catalogs.fda_catalog.models import FDALotProduction
items = db.execute(
select(FDALotProduction).where(
(FDALotProduction.fda_catalog_id == fda_catalog_id) &
(FDALotProduction.tenant_id == tenant_id) &
(FDALotProduction.company_id == company_id)
).order_by(FDALotProduction.line.asc())
).scalars().all()
return items
@staticmethod
def create_lot_production(db: Session, tenant_id: int, company_id: int, fda_catalog_id: int, data: dict):
from api.v1.modules.a76.general_catalogs.fda_catalog.models import FDALotProduction
entry = FDALotProduction(
tenant_id=tenant_id,
company_id=company_id,
fda_catalog_id=fda_catalog_id,
**data
)
db.add(entry)
db.commit()
db.refresh(entry)
return entry
@staticmethod
def delete_lot_production(db: Session, tenant_id: int, company_id: int, lot_id: int):
from api.v1.modules.a76.general_catalogs.fda_catalog.models import FDALotProduction
entry = db.execute(select(FDALotProduction).where(
(FDALotProduction.id == lot_id) &
(FDALotProduction.tenant_id == tenant_id) &
(FDALotProduction.company_id == company_id)
)).scalar_one_or_none()
if entry:
db.delete(entry)
db.commit()
return True
return False