Nuevo formulario de extension para partes SCAI con nuevas tablas de aphis con catalogos fijos nuevos
This commit is contained in:
@@ -0,0 +1,8 @@
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
|
||||
|
||||
class LicenseExceptionDTO(BaseModel):
|
||||
key: str = Field(..., min_length=1, max_length=10)
|
||||
description: str = Field(..., min_length=1, max_length=500)
|
||||
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
@@ -0,0 +1,20 @@
|
||||
from core.database import Base
|
||||
from sqlalchemy import PrimaryKeyConstraint, String
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
|
||||
class LicenseException(Base):
|
||||
__tablename__ = "license_exceptions"
|
||||
__table_args__ = (
|
||||
PrimaryKeyConstraint("key", name="license_exceptions_pkey"),
|
||||
{"schema": "public", "extend_existing": True},
|
||||
)
|
||||
|
||||
key: Mapped[str] = mapped_column(
|
||||
String(10), primary_key=True, nullable=False) # clave del simbolo
|
||||
description: Mapped[str] = mapped_column(
|
||||
String(500), nullable=False
|
||||
) # descripción oficial (en español)
|
||||
|
||||
def __repr__(self):
|
||||
return f"<LicenseException(key={self.key}, description={self.description})>"
|
||||
@@ -0,0 +1,103 @@
|
||||
from typing import Any, Dict
|
||||
|
||||
from core.database import get_core_db
|
||||
from core.security import get_current_user, has_role
|
||||
from fastapi import APIRouter, Depends, HTTPException, Query
|
||||
from sqlalchemy import or_
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from .dto import LicenseExceptionDTO
|
||||
from .models import LicenseException
|
||||
|
||||
router = APIRouter(prefix="/license-exceptions")
|
||||
|
||||
|
||||
@router.get("/", response_model=Dict[str, Any])
|
||||
async def list_license_exceptions(
|
||||
page: int = Query(1, ge=1, description="Número de página"),
|
||||
page_size: int = Query(50, ge=1, le=1000, description="Tamaño de página"),
|
||||
q: str = Query(None, description="Búsqueda general"),
|
||||
db: Session = Depends(get_core_db),
|
||||
current_user: dict = Depends(get_current_user),
|
||||
):
|
||||
skip = (page - 1) * page_size
|
||||
query = db.query(LicenseException)
|
||||
|
||||
if q:
|
||||
search_term = f"%{q}%"
|
||||
query = query.filter(
|
||||
or_(
|
||||
LicenseException.key.ilike(search_term),
|
||||
LicenseException.description.ilike(search_term)
|
||||
)
|
||||
)
|
||||
|
||||
items = query.offset(skip).limit(page_size).all()
|
||||
total = query.count()
|
||||
return {
|
||||
"items": [LicenseExceptionDTO.model_validate(obj) for obj in items],
|
||||
"total": total,
|
||||
"page": page,
|
||||
"page_size": page_size,
|
||||
}
|
||||
|
||||
|
||||
@router.get("/{key}", response_model=LicenseExceptionDTO)
|
||||
async def get_license_exception(
|
||||
key: str,
|
||||
db: Session = Depends(get_core_db),
|
||||
current_user: dict = Depends(get_current_user),
|
||||
):
|
||||
obj = db.query(LicenseException).filter(LicenseException.key == key).first()
|
||||
if not obj:
|
||||
raise HTTPException(status_code=404, detail="Not found")
|
||||
return obj
|
||||
|
||||
|
||||
@router.post("/", response_model=LicenseExceptionDTO, status_code=201)
|
||||
async def create_license_exception(
|
||||
data: LicenseExceptionDTO,
|
||||
db: Session = Depends(get_core_db),
|
||||
current_user: dict = Depends(has_role("admin")),
|
||||
):
|
||||
# Check if already exists
|
||||
existing = db.query(LicenseException).filter(LicenseException.key == data.key).first()
|
||||
if existing:
|
||||
raise HTTPException(status_code=400, detail="License exception with this key already exists")
|
||||
|
||||
obj = LicenseException(**data.dict())
|
||||
db.add(obj)
|
||||
db.commit()
|
||||
db.refresh(obj)
|
||||
return obj
|
||||
|
||||
|
||||
@router.put("/{key}", response_model=LicenseExceptionDTO)
|
||||
async def update_license_exception(
|
||||
key: str,
|
||||
data: LicenseExceptionDTO,
|
||||
db: Session = Depends(get_core_db),
|
||||
current_user: dict = Depends(has_role("admin")),
|
||||
):
|
||||
obj = db.query(LicenseException).filter(LicenseException.key == key).first()
|
||||
if not obj:
|
||||
raise HTTPException(status_code=404, detail="Not found")
|
||||
for field, value in data.dict().items():
|
||||
setattr(obj, field, value)
|
||||
db.commit()
|
||||
db.refresh(obj)
|
||||
return obj
|
||||
|
||||
|
||||
@router.delete("/{key}", status_code=204)
|
||||
async def delete_license_exception(
|
||||
key: str,
|
||||
db: Session = Depends(get_core_db),
|
||||
current_user: dict = Depends(has_role("admin")),
|
||||
):
|
||||
obj = db.query(LicenseException).filter(LicenseException.key == key).first()
|
||||
if not obj:
|
||||
raise HTTPException(status_code=404, detail="Not found")
|
||||
db.delete(obj)
|
||||
db.commit()
|
||||
return None
|
||||
@@ -0,0 +1,41 @@
|
||||
import logging
|
||||
from sqlalchemy.orm import Session
|
||||
from .models import LicenseException
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
LICENSE_EXCEPTIONS_DATA = [
|
||||
{"key": "NLR", "description": "Licencia no requerida (NLR)."},
|
||||
{"key": "LVS", "description": "Los envíos de valor limitado (LVS)."},
|
||||
{"key": "EGB", "description": "Los envíos a los países los países del Grupo B (EGB)."},
|
||||
{"key": "CIV", "description": "Los usuarios finales Civil (CIV)."},
|
||||
{"key": "TSR", "description": "Tecnología y software a restricciones (TSR)."},
|
||||
{"key": "APP", "description": "Informática (APP)."},
|
||||
{"key": "TMP", "description": "Temporales de las importaciones, exportaciones y reexportaciones (TMP)."},
|
||||
{"key": "RPL", "description": "Mantenimiento y sustitución de piezas y equipos (RPL)."},
|
||||
{"key": "GFT", "description": "Los gobiernos, Orgs. internacionales, las inspecciones internacionales en virtud del Convenio de Armas Químicas, y la Estación Espacial Internacional (GOB). Regalo parcelas y las donaciones humanitarias (GFT)."},
|
||||
{"key": "TSU", "description": "Tecnología y software libre (TSU)."},
|
||||
{"key": "BAG", "description": "Equipaje (BAG)."},
|
||||
{"key": "AVS", "description": "Las aeronaves y buques (AVS)."},
|
||||
{"key": "APR", "description": "Adicional reexportación permisiva (APR)."},
|
||||
{"key": "ENC", "description": "Cifrado de productos, software y tecnología (ENC)."},
|
||||
{"key": "AGR", "description": "Productos básicos agrícolas (AGR)."},
|
||||
{"key": "CCD", "description": "Dispositivos de Comunicaciones del Consumidor (CCD)."},
|
||||
]
|
||||
|
||||
def seed_license_exceptions(db: Session):
|
||||
"""Seed License Exceptions catalog data"""
|
||||
logger.info("Seeding License Exceptions...")
|
||||
for item in LICENSE_EXCEPTIONS_DATA:
|
||||
db_item = db.query(LicenseException).filter(LicenseException.key == item["key"]).first()
|
||||
if not db_item:
|
||||
logger.info(f"Adding license exception: {item['key']}")
|
||||
new_item = LicenseException(**item)
|
||||
db.add(new_item)
|
||||
else:
|
||||
# Update description if it changed
|
||||
if db_item.description != item["description"]:
|
||||
logger.info(f"Updating license exception: {item['key']}")
|
||||
db_item.description = item["description"]
|
||||
db.commit()
|
||||
logger.info("License Exceptions seeding completed.")
|
||||
Reference in New Issue
Block a user