363 lines
13 KiB
Python
363 lines
13 KiB
Python
from typing import Dict, Any
|
|
from api.v1.common.tenant_crud_routes import TenantCRUDRoutes
|
|
from core.database import get_core_db
|
|
from core.security import get_current_user, validate_access_to_resource
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, Path
|
|
from sqlalchemy.orm import Session
|
|
|
|
from . import schemas, services
|
|
from .catalog_service import InvoiceCatalogService
|
|
|
|
# Create main router
|
|
router = APIRouter()
|
|
|
|
@router.get("/invoices/creation-data", response_model=schemas.InvoiceCreationResponse)
|
|
def get_creation_data(
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""Get consolidated data for creating a new invoice"""
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
return InvoiceCatalogService.get_creation_data(db, tenant_id, company_id)
|
|
|
|
@router.get("/invoices/{invoice_id}/edition-data", response_model=schemas.InvoiceEditionResponse)
|
|
def get_edition_data(
|
|
invoice_id: int,
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""Get consolidated data for editing an existing invoice"""
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
data = InvoiceCatalogService.get_edition_data(db, invoice_id, tenant_id, company_id)
|
|
if not data:
|
|
raise HTTPException(status_code=404, detail="Invoice not found")
|
|
return data
|
|
|
|
|
|
# Create CRUD routes for Invoice Header using TenantCRUDRoutes
|
|
invoice_crud = TenantCRUDRoutes(
|
|
service=services.InvoiceService,
|
|
create_schema=schemas.InvoiceHeaderCreate,
|
|
update_schema=schemas.InvoiceHeaderUpdate,
|
|
response_schema=schemas.InvoiceHeaderResponse,
|
|
prefix="/invoices",
|
|
tags=[],
|
|
resource_name="Invoice",
|
|
id_name="invoice_id",
|
|
id_type=int,
|
|
enable_list=False, # Disable auto-list to override with custom filter
|
|
enable_filters=True, # Enable filters for status, operation_type, etc.
|
|
list_permissions=[],
|
|
get_permissions=[],
|
|
create_permissions=[],
|
|
update_permissions=[],
|
|
delete_permissions=[],
|
|
default_page_size=50,
|
|
max_page_size=200,
|
|
)
|
|
|
|
# Include the main CRUD routes
|
|
router.include_router(invoice_crud.router)
|
|
|
|
|
|
@router.get("/invoices/", response_model=schemas.InvoiceHeaderListResponse)
|
|
def list_invoices(
|
|
company_id: int = Query(..., description="Company ID"),
|
|
page: int = Query(1, ge=1, description="Page number"),
|
|
page_size: int = Query(50, ge=1, le=200, description="Items per page"),
|
|
search: str = Query(None, description="Search by invoice number"),
|
|
status: bool = Query(None, description="Filter by status"),
|
|
operation_type: schemas.OperationType = Query(None, description="Filter by operation type"),
|
|
invoice_type: str = Query(None, description="Filter by invoice type"),
|
|
manifest_number: str = Query(None, description="Filter by manifest number"),
|
|
pedimento: str = Query(None, description="Filter by pedimento"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""
|
|
List invoices with optional filters, including manifest_number.
|
|
"""
|
|
print(f"DEBUG: list_invoices called with manifest_number={manifest_number}")
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
print(f"DEBUG: tenant_id={tenant_id}, company_id={company_id}")
|
|
|
|
skip = (page - 1) * page_size
|
|
filters = {
|
|
"invoice_number": search,
|
|
"status": status,
|
|
"operation_type": operation_type,
|
|
"invoice_type": invoice_type,
|
|
"manifest_number": manifest_number,
|
|
"pedimento": pedimento,
|
|
}
|
|
|
|
# Remove None values
|
|
filters = {k: v for k, v in filters.items() if v is not None}
|
|
|
|
items, total = services.InvoiceService.get_all(
|
|
db, tenant_id, company_id, skip=skip, limit=page_size, filters=filters
|
|
)
|
|
print(f"DEBUG: InvoiceService returned {len(items)} items, total={total}")
|
|
|
|
return {
|
|
"items": items,
|
|
"total": total,
|
|
"page": page,
|
|
"page_size": page_size
|
|
}
|
|
|
|
|
|
# Additional nested routes for child resources
|
|
|
|
# --- Logistics Routes ---
|
|
|
|
@router.get(
|
|
"/invoices/{invoice_id}/logistics",
|
|
response_model=list[schemas.InvoiceLogisticsResponse],
|
|
summary="Get all logistics for an invoice",
|
|
)
|
|
def get_invoice_logistics(
|
|
invoice_id: int = Path(..., description="Invoice ID"),
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""Get all logistics entries for a specific invoice"""
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
|
|
# Verify the invoice exists and belongs to the tenant/company
|
|
invoice = services.InvoiceService.get_by_id(
|
|
db, invoice_id, tenant_id, company_id)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail="Invoice not found")
|
|
|
|
logistics = services.InvoiceLogisticsService.get_all_by_invoice(
|
|
db, invoice_id)
|
|
return logistics
|
|
|
|
|
|
@router.post(
|
|
"/invoices/{invoice_id}/logistics",
|
|
response_model=schemas.InvoiceLogisticsResponse,
|
|
status_code=201,
|
|
summary="Add logistics to an invoice",
|
|
)
|
|
def create_invoice_logistics(
|
|
invoice_id: int = Path(..., description="Invoice ID"),
|
|
logistics_data: schemas.InvoiceLogisticsCreate = ...,
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""Add a new logistics entry to an invoice"""
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
|
|
# Verify the invoice exists and belongs to the tenant/company
|
|
invoice = services.InvoiceService.get_by_id(
|
|
db, invoice_id, tenant_id, company_id)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail="Invoice not found")
|
|
|
|
logistics = services.InvoiceLogisticsService.create(
|
|
db, logistics_data, invoice_id, tenant_id, company_id)
|
|
return logistics
|
|
|
|
|
|
@router.delete(
|
|
"/invoices/{invoice_id}/logistics/{logistics_id}",
|
|
status_code=204,
|
|
summary="Delete logistics from an invoice",
|
|
)
|
|
def delete_invoice_logistics(
|
|
invoice_id: int = Path(..., description="Invoice ID"),
|
|
logistics_id: int = Path(..., description="Logistics ID"),
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""Delete a logistics entry from an invoice"""
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
|
|
# Verify the invoice exists and belongs to the tenant/company
|
|
invoice = services.InvoiceService.get_by_id(
|
|
db, invoice_id, tenant_id, company_id)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail="Invoice not found")
|
|
|
|
success = services.InvoiceLogisticsService.delete(
|
|
db, logistics_id, invoice_id)
|
|
if not success:
|
|
raise HTTPException(
|
|
status_code=404, detail="Logistics entry not found")
|
|
|
|
return None
|
|
|
|
|
|
# --- Sales Details Routes ---
|
|
|
|
@router.get(
|
|
"/invoices/{invoice_id}/details",
|
|
response_model=list[schemas.InvoiceSalesDetailsResponse],
|
|
summary="Get all sales details for an invoice",
|
|
)
|
|
def get_invoice_details(
|
|
invoice_id: int = Path(..., description="Invoice ID"),
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""Get all sales details for a specific invoice"""
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
|
|
# Verify the invoice exists and belongs to the tenant/company
|
|
invoice = services.InvoiceService.get_by_id(
|
|
db, invoice_id, tenant_id, company_id)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail="Invoice not found")
|
|
|
|
details = services.InvoiceSalesDetailsService.get_all_by_invoice(
|
|
db, invoice_id)
|
|
return details
|
|
|
|
|
|
@router.post(
|
|
"/invoices/{invoice_id}/details",
|
|
response_model=schemas.InvoiceSalesDetailsResponse,
|
|
status_code=201,
|
|
summary="Add sales detail to an invoice",
|
|
)
|
|
|
|
def create_invoice_detail(
|
|
invoice_id: int = Path(..., description="Invoice ID"),
|
|
detail_data: schemas.InvoiceSalesDetailsCreate = ...,
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""Add a new sales detail to an invoice"""
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
|
|
# Verify the invoice exists and belongs to the tenant/company
|
|
invoice = services.InvoiceService.get_by_id(
|
|
db, invoice_id, tenant_id, company_id)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail="Invoice not found")
|
|
|
|
detail = services.InvoiceSalesDetailsService.create(
|
|
db, detail_data, invoice_id, tenant_id, company_id)
|
|
return detail
|
|
|
|
|
|
@router.delete(
|
|
"/invoices/{invoice_id}/details/{detail_id}",
|
|
status_code=204,
|
|
summary="Delete sales detail from an invoice",
|
|
)
|
|
def delete_invoice_detail(
|
|
invoice_id: int = Path(..., description="Invoice ID"),
|
|
detail_id: int = Path(..., description="Detail ID"),
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""Delete a sales detail from an invoice"""
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
|
|
# Verify the invoice exists and belongs to the tenant/company
|
|
invoice = services.InvoiceService.get_by_id(
|
|
db, invoice_id, tenant_id, company_id)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail="Invoice not found")
|
|
|
|
success = services.InvoiceSalesDetailsService.delete(
|
|
db, detail_id, invoice_id)
|
|
if not success:
|
|
raise HTTPException(status_code=404, detail="Sales detail not found")
|
|
|
|
return None
|
|
|
|
|
|
# --- Collections Routes ---
|
|
|
|
@router.get(
|
|
"/invoices/{invoice_id}/collections",
|
|
response_model=list[schemas.InvoiceCollectionsResponse],
|
|
summary="Get all collections for an invoice",
|
|
)
|
|
def get_invoice_collections(
|
|
invoice_id: int = Path(..., description="Invoice ID"),
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""Get all collections for a specific invoice"""
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
|
|
# Verify the invoice exists and belongs to the tenant/company
|
|
invoice = services.InvoiceService.get_by_id(
|
|
db, invoice_id, tenant_id, company_id)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail="Invoice not found")
|
|
|
|
collections = services.InvoiceCollectionsService.get_all_by_invoice(
|
|
db, invoice_id)
|
|
return collections
|
|
|
|
|
|
@router.post(
|
|
"/invoices/{invoice_id}/collections",
|
|
response_model=schemas.InvoiceCollectionsResponse,
|
|
status_code=201,
|
|
summary="Add collection to an invoice",
|
|
)
|
|
def create_invoice_collection(
|
|
invoice_id: int = Path(..., description="Invoice ID"),
|
|
collection_data: schemas.InvoiceCollectionsCreate = ...,
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""Add a new collection to an invoice"""
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
|
|
# Verify the invoice exists and belongs to the tenant/company
|
|
invoice = services.InvoiceService.get_by_id(
|
|
db, invoice_id, tenant_id, company_id)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail="Invoice not found")
|
|
|
|
collection = services.InvoiceCollectionsService.create(
|
|
db, collection_data, invoice_id, tenant_id, company_id)
|
|
return collection
|
|
|
|
|
|
@router.delete(
|
|
"/invoices/{invoice_id}/collections/{collection_id}",
|
|
status_code=204,
|
|
summary="Delete collection from an invoice",
|
|
)
|
|
def delete_invoice_collection(
|
|
invoice_id: int = Path(..., description="Invoice ID"),
|
|
collection_id: int = Path(..., description="Collection ID"),
|
|
company_id: int = Query(..., description="Company ID"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""Delete a collection from an invoice"""
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
|
|
|
# Verify the invoice exists and belongs to the tenant/company
|
|
invoice = services.InvoiceService.get_by_id(
|
|
db, invoice_id, tenant_id, company_id)
|
|
if not invoice:
|
|
raise HTTPException(status_code=404, detail="Invoice not found")
|
|
|
|
success = services.InvoiceCollectionsService.delete(
|
|
db, collection_id, invoice_id)
|
|
if not success:
|
|
raise HTTPException(status_code=404, detail="Collection not found")
|
|
|
|
return None
|