290 lines
10 KiB
Python
290 lines
10 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
|
|
|
|
# Create main router
|
|
router = APIRouter()
|
|
|
|
# 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=True, # Enable list endpoint with pagination
|
|
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)
|
|
|
|
|
|
# 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
|