0b12ad5354
- schema fin: fin.invoices + fin.invoice_items + fin.payments; totales con IVA, estados borrador→emitida→enviada→pagada, cobranza (pagos) y saldo automático - generar-factura-desde-embarque (toma conceptos de venta de la cotización) - ops.shipment_events: bitácora/hitos del embarque con secuencia por defecto según operación (importación/exportación) — cubre Diagrama 3 - subida de documentos a MinIO: POST /crm/uploads (multipart) + URL firmada - migración c4d5e6f7a8b9, routers/permisos (fin), seed del flujo hasta factura - 58 tests pytest en verde Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
115 lines
5.9 KiB
Python
115 lines
5.9 KiB
Python
from fastapi import APIRouter, Depends, Query, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from core.database import get_core_db
|
|
from core.security import get_current_user
|
|
|
|
from . import service
|
|
from .dto import (
|
|
InvoiceCreate,
|
|
InvoiceItemCreate,
|
|
InvoiceItemResponse,
|
|
InvoiceItemUpdate,
|
|
InvoiceResponse,
|
|
InvoiceUpdate,
|
|
PaymentCreate,
|
|
PaymentResponse,
|
|
)
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def _uid(cu: dict) -> str | None:
|
|
return cu.get("sub") or cu.get("id")
|
|
|
|
|
|
@router.get("/invoices", response_model=list[InvoiceResponse])
|
|
def list_invoices(
|
|
company_id: int = Query(...),
|
|
search: str | None = Query(None),
|
|
inv_status: str | None = Query(None, alias="status"),
|
|
account_id: int | None = Query(None),
|
|
current_user: dict = Depends(get_current_user),
|
|
db: Session = Depends(get_core_db),
|
|
):
|
|
return service.get_invoices(db, current_user["tenant_id"], company_id, search, inv_status, account_id)
|
|
|
|
|
|
@router.get("/invoices/{invoice_id}", response_model=InvoiceResponse)
|
|
def get_invoice(invoice_id: int, company_id: int = Query(...), current_user: dict = Depends(get_current_user), db: Session = Depends(get_core_db)):
|
|
return service.get_invoice(db, invoice_id, current_user["tenant_id"], company_id)
|
|
|
|
|
|
@router.post("/invoices", response_model=InvoiceResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_invoice(payload: InvoiceCreate, company_id: int = Query(...), current_user: dict = Depends(get_current_user), db: Session = Depends(get_core_db)):
|
|
return service.create_invoice(db, payload, current_user["tenant_id"], company_id, _uid(current_user))
|
|
|
|
|
|
@router.post("/invoices/from-shipment", response_model=InvoiceResponse, status_code=status.HTTP_201_CREATED)
|
|
def generate_from_shipment(shipment_id: int = Query(...), company_id: int = Query(...), current_user: dict = Depends(get_current_user), db: Session = Depends(get_core_db)):
|
|
return service.generate_from_shipment(db, shipment_id, current_user["tenant_id"], company_id, _uid(current_user))
|
|
|
|
|
|
@router.patch("/invoices/{invoice_id}", response_model=InvoiceResponse)
|
|
def update_invoice(invoice_id: int, payload: InvoiceUpdate, company_id: int = Query(...), current_user: dict = Depends(get_current_user), db: Session = Depends(get_core_db)):
|
|
return service.update_invoice(db, invoice_id, payload, current_user["tenant_id"], company_id, _uid(current_user))
|
|
|
|
|
|
@router.patch("/invoices/{invoice_id}/emit", response_model=InvoiceResponse)
|
|
def emit_invoice(invoice_id: int, company_id: int = Query(...), current_user: dict = Depends(get_current_user), db: Session = Depends(get_core_db)):
|
|
return service.emit_invoice(db, invoice_id, current_user["tenant_id"], company_id)
|
|
|
|
|
|
@router.patch("/invoices/{invoice_id}/send", response_model=InvoiceResponse)
|
|
def send_invoice(invoice_id: int, company_id: int = Query(...), current_user: dict = Depends(get_current_user), db: Session = Depends(get_core_db)):
|
|
return service.send_invoice(db, invoice_id, current_user["tenant_id"], company_id)
|
|
|
|
|
|
@router.patch("/invoices/{invoice_id}/cancel", response_model=InvoiceResponse)
|
|
def cancel_invoice(invoice_id: int, company_id: int = Query(...), current_user: dict = Depends(get_current_user), db: Session = Depends(get_core_db)):
|
|
return service.cancel_invoice(db, invoice_id, current_user["tenant_id"], company_id)
|
|
|
|
|
|
@router.delete("/invoices/{invoice_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_invoice(invoice_id: int, company_id: int = Query(...), current_user: dict = Depends(get_current_user), db: Session = Depends(get_core_db)):
|
|
service.delete_invoice(db, invoice_id, current_user["tenant_id"], company_id)
|
|
|
|
|
|
# ----- Conceptos -----
|
|
|
|
@router.get("/invoices/{invoice_id}/items", response_model=list[InvoiceItemResponse])
|
|
def list_items(invoice_id: int, company_id: int = Query(...), current_user: dict = Depends(get_current_user), db: Session = Depends(get_core_db)):
|
|
return service.get_items(db, invoice_id, current_user["tenant_id"], company_id)
|
|
|
|
|
|
@router.post("/invoice-items", response_model=InvoiceItemResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_item(payload: InvoiceItemCreate, company_id: int = Query(...), current_user: dict = Depends(get_current_user), db: Session = Depends(get_core_db)):
|
|
return service.create_item(db, payload, current_user["tenant_id"], company_id)
|
|
|
|
|
|
@router.patch("/invoice-items/{item_id}", response_model=InvoiceItemResponse)
|
|
def update_item(item_id: int, payload: InvoiceItemUpdate, company_id: int = Query(...), current_user: dict = Depends(get_current_user), db: Session = Depends(get_core_db)):
|
|
return service.update_item(db, item_id, payload, current_user["tenant_id"], company_id)
|
|
|
|
|
|
@router.delete("/invoice-items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_item(item_id: int, company_id: int = Query(...), current_user: dict = Depends(get_current_user), db: Session = Depends(get_core_db)):
|
|
service.delete_item(db, item_id, current_user["tenant_id"], company_id)
|
|
|
|
|
|
# ----- Pagos (cobranza) -----
|
|
|
|
@router.get("/invoices/{invoice_id}/payments", response_model=list[PaymentResponse])
|
|
def list_payments(invoice_id: int, company_id: int = Query(...), current_user: dict = Depends(get_current_user), db: Session = Depends(get_core_db)):
|
|
return service.get_payments(db, invoice_id, current_user["tenant_id"], company_id)
|
|
|
|
|
|
@router.post("/payments", response_model=PaymentResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_payment(payload: PaymentCreate, company_id: int = Query(...), current_user: dict = Depends(get_current_user), db: Session = Depends(get_core_db)):
|
|
return service.create_payment(db, payload, current_user["tenant_id"], company_id)
|
|
|
|
|
|
@router.delete("/payments/{payment_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_payment(payment_id: int, company_id: int = Query(...), current_user: dict = Depends(get_current_user), db: Session = Depends(get_core_db)):
|
|
service.delete_payment(db, payment_id, current_user["tenant_id"], company_id)
|