Merge branch 'feature/tables-generation' into development

This commit is contained in:
2026-02-11 17:26:09 -06:00
23 changed files with 3451 additions and 483 deletions

View File

@@ -0,0 +1,28 @@
from typing import Any, Dict, Optional
from pydantic import BaseModel, ConfigDict
from enum import Enum
class OperationType(str, Enum):
IMP = "imp" # Importación
EXP = "exp" # Exportación
SM_IN = "sm_in" # Entrada SM
SM_OUT = "sm_out" # Salida SM
CTM_SEND = "ctm_send" # Envío CTM
CTM_RECEIVE = "ctm_receive" # Recibo CTM
class InvoiceSettingsBase(BaseModel):
invoice_type: str
operation_type: OperationType
settings: Dict[str, Any]
class InvoiceSettingsRequest(InvoiceSettingsBase):
"""Schema for creating or updating invoice settings"""
pass
class InvoiceSettingsResponse(InvoiceSettingsBase):
"""Schema for returning invoice settings"""
id: int
tenant_id: int
company_id: int
model_config = ConfigDict(from_attributes=True)

View File

@@ -0,0 +1,47 @@
from typing import Optional
from sqlalchemy import BigInteger, ForeignKey, String, UniqueConstraint
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column
from core.database import Base
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
import enum
class OperationType(str, enum.Enum):
IMP = "imp" # Importación
EXP = "exp" # Exportación
SM_IN = "sm_in" # Entrada SM
SM_OUT = "sm_out" # Salida SM
CTM_SEND = "ctm_send" # Envío CTM
CTM_RECEIVE = "ctm_receive" # Recibo CTM
class InvoiceSettings(Base, TenantScopedMixin, TimestampMixin):
__tablename__ = "invoice_settings"
__table_args__ = (
UniqueConstraint(
"tenant_id",
"company_id",
"invoice_type",
"operation_type",
name="uq_invoice_settings_tenant_company_type_op"
),
{"schema": "a76"},
)
id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
# Configuration Scope
invoice_type: Mapped[str] = mapped_column(
ForeignKey("public.invoice_types.key"),
nullable=False
)
operation_type: Mapped[OperationType] = mapped_column(
String(11),
nullable=False
)
# The actual settings payload
settings: Mapped[dict] = mapped_column(JSONB, nullable=False, default={})
def __repr__(self):
return f"<InvoiceSettings(id={self.id}, type={self.invoice_type}, op={self.operation_type})>"

View File

@@ -0,0 +1,76 @@
from typing import List, Dict, Any
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from core.database import get_core_db
from core.security import get_current_user, validate_access_to_resource
from api.v1.modules.a76.invoice_settings import services
from api.v1.modules.a76.invoice_settings.dto import InvoiceSettingsRequest, InvoiceSettingsResponse, OperationType
router = APIRouter(
prefix="/a76/invoice-settings",
tags=["a76/invoice-settings"]
)
@router.get("/{invoice_type}", response_model=InvoiceSettingsResponse)
def get_invoice_settings(
invoice_type: str,
operation_type: OperationType = Query(...),
company_id: int = Query(...),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""Get settings for a specific invoice type and operation"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
settings = services.get_settings(
db,
tenant_id,
company_id,
invoice_type,
operation_type
)
if not settings:
# Return empty default if not found, to simplify frontend logic
return InvoiceSettingsResponse(
invoice_type=invoice_type,
operation_type=operation_type,
settings={},
id=0,
tenant_id=tenant_id,
company_id=company_id
)
return settings
@router.get("/", response_model=List[InvoiceSettingsResponse])
def list_invoice_settings(
company_id: int = Query(...),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""List all configured settings for validation or overview"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
return services.list_settings(
db,
tenant_id,
company_id
)
@router.put("/", response_model=InvoiceSettingsResponse)
def save_invoice_settings(
settings_data: InvoiceSettingsRequest,
company_id: int = Query(...),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""Create or update invoice settings"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
return services.upsert_settings(
db,
tenant_id,
company_id,
settings_data
)

View File

@@ -0,0 +1,70 @@
from typing import List, Optional
from sqlalchemy.orm import Session
from sqlalchemy import select
from fastapi import HTTPException
from api.v1.modules.a76.invoice_settings.models import InvoiceSettings
from api.v1.modules.a76.invoice_settings.dto import InvoiceSettingsRequest, OperationType
def get_settings(
db: Session,
tenant_id: int,
company_id: int,
invoice_type: str,
operation_type: OperationType
) -> Optional[InvoiceSettings]:
"""Retrieve settings for a specific context"""
stmt = select(InvoiceSettings).where(
InvoiceSettings.tenant_id == tenant_id,
InvoiceSettings.company_id == company_id,
InvoiceSettings.invoice_type == invoice_type,
InvoiceSettings.operation_type == operation_type
)
return db.execute(stmt).scalar_one_or_none()
def list_settings(
db: Session,
tenant_id: int,
company_id: int
) -> List[InvoiceSettings]:
"""List all settings for a company"""
stmt = select(InvoiceSettings).where(
InvoiceSettings.tenant_id == tenant_id,
InvoiceSettings.company_id == company_id
)
return db.execute(stmt).scalars().all()
def upsert_settings(
db: Session,
tenant_id: int,
company_id: int,
settings_data: InvoiceSettingsRequest
) -> InvoiceSettings:
"""Create or update settings"""
# Check if exists
existing = get_settings(
db,
tenant_id,
company_id,
settings_data.invoice_type,
settings_data.operation_type
)
if existing:
existing.settings = settings_data.settings
db.commit()
db.refresh(existing)
return existing
# Create new
new_settings = InvoiceSettings(
tenant_id=tenant_id,
company_id=company_id,
invoice_type=settings_data.invoice_type,
operation_type=settings_data.operation_type,
settings=settings_data.settings
)
db.add(new_settings)
db.commit()
db.refresh(new_settings)
return new_settings

View File

@@ -0,0 +1,25 @@
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, ConfigDict
class ItemPresetBase(BaseModel):
name: str
description: Optional[str] = None
items: List[Dict[str, Any]]
class ItemPresetCreate(ItemPresetBase):
"""Schema for creating a new item preset"""
pass
class ItemPresetUpdate(BaseModel):
"""Schema for updating an existing item preset"""
name: Optional[str] = None
description: Optional[str] = None
items: Optional[List[Dict[str, Any]]] = None
class ItemPresetResponse(ItemPresetBase):
"""Schema for returning an item preset"""
id: int
tenant_id: int
company_id: int
model_config = ConfigDict(from_attributes=True)

View File

@@ -0,0 +1,18 @@
from typing import Any, Dict, List
from sqlalchemy import String, Integer, JSON
from sqlalchemy.orm import Mapped, mapped_column
from core.database import Base
from api.v1.common.base_models import TenantScopedMixin, TimestampMixin
class ItemPreset(Base, TenantScopedMixin, TimestampMixin):
__tablename__ = "item_presets"
__table_args__ = {"schema": "a76"}
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
name: Mapped[str] = mapped_column(String(100), nullable=False)
description: Mapped[str | None] = mapped_column(String(500), nullable=True)
items: Mapped[List[Dict[str, Any]]] = mapped_column(JSON, nullable=False, default=[])
def __repr__(self):
return f"<ItemPreset(id={self.id}, name='{self.name}')>"

View File

@@ -0,0 +1,72 @@
from typing import List, Any, Dict
from fastapi import APIRouter, Depends, Query, HTTPException
from sqlalchemy.orm import Session
from core.database import get_core_db
from core.security import get_current_user, validate_access_to_resource
from . import services, dto
router = APIRouter()
@router.get("/", response_model=List[dto.ItemPresetResponse])
def list_item_presets(
company_id: int = Query(...),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""List all item presets for a company"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
return services.get_presets(db, tenant_id, company_id)
@router.post("/", response_model=dto.ItemPresetResponse)
def create_item_preset(
data: dto.ItemPresetCreate,
company_id: int = Query(...),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""Create a new item preset"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
return services.create_preset(db, tenant_id, company_id, data)
@router.get("/{preset_id}", response_model=dto.ItemPresetResponse)
def get_item_preset(
preset_id: int,
company_id: int = Query(...),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""Get a specific item preset"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
preset = services.get_preset_by_id(db, preset_id, tenant_id, company_id)
if not preset:
raise HTTPException(status_code=404, detail="Item preset not found")
return preset
@router.put("/{preset_id}", response_model=dto.ItemPresetResponse)
def update_item_preset(
preset_id: int,
data: dto.ItemPresetUpdate,
company_id: int = Query(...),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""Update an item preset"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
preset = services.update_preset(db, preset_id, tenant_id, company_id, data)
if not preset:
raise HTTPException(status_code=404, detail="Item preset not found")
return preset
@router.delete("/{preset_id}")
def delete_item_preset(
preset_id: int,
company_id: int = Query(...),
db: Session = Depends(get_core_db),
current_user: Dict[str, Any] = Depends(get_current_user),
):
"""Delete an item preset"""
tenant_id = validate_access_to_resource(db, company_id, current_user)
success = services.delete_preset(db, preset_id, tenant_id, company_id)
if not success:
raise HTTPException(status_code=404, detail="Item preset not found")
return {"status": "success", "message": "Preset deleted"}

View File

@@ -0,0 +1,63 @@
from typing import List, Optional
from sqlalchemy.orm import Session
from .models import ItemPreset
from .dto import ItemPresetCreate, ItemPresetUpdate
def get_presets(db: Session, tenant_id: int, company_id: int) -> List[ItemPreset]:
"""Get all presets for a specific company and tenant"""
return db.query(ItemPreset).filter(
ItemPreset.tenant_id == tenant_id,
ItemPreset.company_id == company_id
).all()
def get_preset_by_id(db: Session, preset_id: int, tenant_id: int, company_id: int) -> Optional[ItemPreset]:
"""Get a specific preset by ID"""
return db.query(ItemPreset).filter(
ItemPreset.id == preset_id,
ItemPreset.tenant_id == tenant_id,
ItemPreset.company_id == company_id
).first()
def create_preset(db: Session, tenant_id: int, company_id: int, data: ItemPresetCreate) -> ItemPreset:
"""Create a new item preset"""
new_preset = ItemPreset(
tenant_id=tenant_id,
company_id=company_id,
name=data.name,
description=data.description,
items=data.items
)
db.add(new_preset)
db.commit()
db.refresh(new_preset)
return new_preset
def update_preset(
db: Session,
preset_id: int,
tenant_id: int,
company_id: int,
data: ItemPresetUpdate
) -> Optional[ItemPreset]:
"""Update an existing item preset"""
preset = get_preset_by_id(db, preset_id, tenant_id, company_id)
if not preset:
return None
update_data = data.model_dump(exclude_unset=True)
for key, value in update_data.items():
setattr(preset, key, value)
db.commit()
db.refresh(preset)
return preset
def delete_preset(db: Session, preset_id: int, tenant_id: int, company_id: int) -> bool:
"""Delete an item preset"""
preset = get_preset_by_id(db, preset_id, tenant_id, company_id)
if not preset:
return False
db.delete(preset)
db.commit()
return True

View File

@@ -12,6 +12,8 @@ from .invoices.routes import router as invoices_router
from .items.routes import router as items_router
from .classes import router as classes_router
from .clients_and_providers import router as client_and_provider_router
from .invoice_settings.routes import router as invoice_settings_router
from .item_presets.routes import router as item_presets_router
from .general_catalogs.company import router as company_router
from .country_rule_oct.routes import router as country_rule_oct_router
from .transportation.drivers.routes import router as drivers_router
@@ -66,6 +68,8 @@ router = APIRouter()
# Registrar módulos
router.include_router(invoices_router, prefix="/a76", tags=["a76 / invoices"])
router.include_router(items_router, prefix="/a76", tags=["a76 / items"])
router.include_router(invoice_settings_router)
router.include_router(item_presets_router, prefix="/a76/item-presets", tags=["a76 / item_presets"])
router.include_router(pedimentos_router, prefix="/a76")
router.include_router(
client_and_provider_router, prefix="/a76", tags=["a76 / clients_and_providers"]