feat: Implement invoice management module with CRUD operations
- Added InvoiceHeader, InvoiceComplianceMx, InvoiceFinancials, InvoiceLogistics, InvoiceSalesDetails, and InvoiceCollections models. - Created schemas for invoice operations including create, update, and response schemas. - Developed services for handling business logic related to invoices, including retrieval, creation, updating, and deletion of invoices and their related data. - Introduced routes for invoice management, enabling CRUD operations through a RESTful API. - Integrated invoice routes into the main application router. - Removed unused routers from the core module to streamline the API structure.
This commit is contained in:
369
backend/api/v1/modules/a76/invoices/services.py
Normal file
369
backend/api/v1/modules/a76/invoices/services.py
Normal file
@@ -0,0 +1,369 @@
|
||||
from typing import Optional, List, Tuple
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlalchemy import and_
|
||||
|
||||
from . import models, schemas
|
||||
|
||||
|
||||
class InvoiceService:
|
||||
"""Service for Invoice Header operations"""
|
||||
|
||||
@staticmethod
|
||||
def get_by_id(db: Session, invoice_id: int, tenant_id: int, company_id: int) -> Optional[models.InvoiceHeader]:
|
||||
"""Get an invoice by ID with tenant/company validation"""
|
||||
return (
|
||||
db.query(models.InvoiceHeader)
|
||||
.filter(
|
||||
models.InvoiceHeader.id == invoice_id,
|
||||
models.InvoiceHeader.tenant_id == tenant_id,
|
||||
models.InvoiceHeader.company_id == company_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def get_all(
|
||||
db: Session,
|
||||
tenant_id: int,
|
||||
company_id: int,
|
||||
skip: int = 0,
|
||||
limit: int = 100,
|
||||
filters: Optional[dict] = None,
|
||||
) -> Tuple[List[models.InvoiceHeader], int]:
|
||||
"""Get all invoices for a tenant/company with pagination and optional filters"""
|
||||
query = db.query(models.InvoiceHeader).filter(
|
||||
models.InvoiceHeader.tenant_id == tenant_id,
|
||||
models.InvoiceHeader.company_id == company_id,
|
||||
)
|
||||
|
||||
# Apply filters if provided
|
||||
if filters:
|
||||
if filters.get("status"):
|
||||
query = query.filter(
|
||||
models.InvoiceHeader.status == filters["status"])
|
||||
if filters.get("operation_type"):
|
||||
query = query.filter(
|
||||
models.InvoiceHeader.operation_type == filters["operation_type"])
|
||||
if filters.get("invoice_number"):
|
||||
query = query.filter(models.InvoiceHeader.invoice_number.ilike(
|
||||
f"%{filters['invoice_number']}%"))
|
||||
if filters.get("pedimento"):
|
||||
query = query.join(models.InvoiceComplianceMx).filter(
|
||||
models.InvoiceComplianceMx.pedimento.ilike(
|
||||
f"%{filters['pedimento']}%")
|
||||
)
|
||||
|
||||
total = query.count()
|
||||
items = query.offset(skip).limit(limit).all()
|
||||
return items, total
|
||||
|
||||
@staticmethod
|
||||
def create(
|
||||
db: Session,
|
||||
invoice_data: schemas.InvoiceHeaderCreate,
|
||||
tenant_id: int,
|
||||
company_id: int
|
||||
) -> models.InvoiceHeader:
|
||||
"""Create a new invoice with all related data"""
|
||||
# Extract nested data
|
||||
compliance_data = invoice_data.compliance_mx
|
||||
financials_data = invoice_data.financials
|
||||
logistics_data = invoice_data.logistics or []
|
||||
details_data = invoice_data.details or []
|
||||
collections_data = invoice_data.collections or []
|
||||
|
||||
# Create main invoice header
|
||||
invoice_dict = invoice_data.model_dump(
|
||||
exclude={"compliance_mx", "financials",
|
||||
"logistics", "details", "collections"}
|
||||
)
|
||||
invoice_dict["tenant_id"] = tenant_id
|
||||
invoice_dict["company_id"] = company_id
|
||||
|
||||
new_invoice = models.InvoiceHeader(**invoice_dict)
|
||||
db.add(new_invoice)
|
||||
db.flush() # Flush to get the invoice ID
|
||||
|
||||
# Create compliance_mx if provided
|
||||
if compliance_data:
|
||||
compliance_dict = compliance_data.model_dump()
|
||||
compliance_dict["invoice_id"] = new_invoice.id
|
||||
compliance_dict["tenant_id"] = tenant_id
|
||||
compliance_dict["company_id"] = company_id
|
||||
new_compliance = models.InvoiceComplianceMx(**compliance_dict)
|
||||
db.add(new_compliance)
|
||||
|
||||
# Create financials if provided
|
||||
if financials_data:
|
||||
financials_dict = financials_data.model_dump()
|
||||
financials_dict["invoice_id"] = new_invoice.id
|
||||
financials_dict["tenant_id"] = tenant_id
|
||||
financials_dict["company_id"] = company_id
|
||||
new_financials = models.InvoiceFinancials(**financials_dict)
|
||||
db.add(new_financials)
|
||||
|
||||
# Create logistics entries
|
||||
for logistics_item in logistics_data:
|
||||
logistics_dict = logistics_item.model_dump()
|
||||
logistics_dict["invoice_id"] = new_invoice.id
|
||||
logistics_dict["tenant_id"] = tenant_id
|
||||
logistics_dict["company_id"] = company_id
|
||||
new_logistics = models.InvoiceLogistics(**logistics_dict)
|
||||
db.add(new_logistics)
|
||||
|
||||
# Create sales details
|
||||
for detail_item in details_data:
|
||||
detail_dict = detail_item.model_dump()
|
||||
detail_dict["invoice_id"] = new_invoice.id
|
||||
detail_dict["tenant_id"] = tenant_id
|
||||
detail_dict["company_id"] = company_id
|
||||
new_detail = models.InvoiceSalesDetails(**detail_dict)
|
||||
db.add(new_detail)
|
||||
|
||||
# Create collections
|
||||
for collection_item in collections_data:
|
||||
collection_dict = collection_item.model_dump()
|
||||
collection_dict["invoice_id"] = new_invoice.id
|
||||
collection_dict["tenant_id"] = tenant_id
|
||||
collection_dict["company_id"] = company_id
|
||||
new_collection = models.InvoiceCollections(**collection_dict)
|
||||
db.add(new_collection)
|
||||
|
||||
db.commit()
|
||||
db.refresh(new_invoice)
|
||||
return new_invoice
|
||||
|
||||
@staticmethod
|
||||
def update(
|
||||
db: Session,
|
||||
invoice_id: int,
|
||||
tenant_id: int,
|
||||
invoice_data: schemas.InvoiceHeaderUpdate,
|
||||
company_id: int
|
||||
) -> Optional[models.InvoiceHeader]:
|
||||
"""Update an existing invoice and its related data"""
|
||||
invoice = InvoiceService.get_by_id(
|
||||
db, invoice_id, tenant_id, company_id)
|
||||
if not invoice:
|
||||
return None
|
||||
|
||||
# Update main invoice header fields
|
||||
update_dict = invoice_data.model_dump(
|
||||
exclude={"compliance_mx", "financials",
|
||||
"logistics", "details", "collections"},
|
||||
exclude_unset=True
|
||||
)
|
||||
for key, value in update_dict.items():
|
||||
setattr(invoice, key, value)
|
||||
|
||||
# Update compliance_mx if provided
|
||||
if invoice_data.compliance_mx is not None:
|
||||
if invoice.compliance_mx:
|
||||
for key, value in invoice_data.compliance_mx.model_dump(exclude_unset=True).items():
|
||||
setattr(invoice.compliance_mx, key, value)
|
||||
else:
|
||||
compliance_dict = invoice_data.compliance_mx.model_dump()
|
||||
compliance_dict["invoice_id"] = invoice.id
|
||||
compliance_dict["tenant_id"] = tenant_id
|
||||
compliance_dict["company_id"] = company_id
|
||||
new_compliance = models.InvoiceComplianceMx(**compliance_dict)
|
||||
db.add(new_compliance)
|
||||
|
||||
# Update financials if provided
|
||||
if invoice_data.financials is not None:
|
||||
if invoice.financials:
|
||||
for key, value in invoice_data.financials.model_dump(exclude_unset=True).items():
|
||||
setattr(invoice.financials, key, value)
|
||||
else:
|
||||
financials_dict = invoice_data.financials.model_dump()
|
||||
financials_dict["invoice_id"] = invoice.id
|
||||
financials_dict["tenant_id"] = tenant_id
|
||||
financials_dict["company_id"] = company_id
|
||||
new_financials = models.InvoiceFinancials(**financials_dict)
|
||||
db.add(new_financials)
|
||||
|
||||
# Note: For logistics, details, and collections, we're not handling updates here
|
||||
# as they are typically managed through separate endpoints for complex operations
|
||||
|
||||
db.commit()
|
||||
db.refresh(invoice)
|
||||
return invoice
|
||||
|
||||
@staticmethod
|
||||
def delete(db: Session, invoice_id: int, tenant_id: int, company_id: int) -> bool:
|
||||
"""Delete an invoice and all related data (cascade delete)"""
|
||||
invoice = InvoiceService.get_by_id(
|
||||
db, invoice_id, tenant_id, company_id)
|
||||
if invoice:
|
||||
db.delete(invoice)
|
||||
db.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
class InvoiceLogisticsService:
|
||||
"""Service for Invoice Logistics operations"""
|
||||
|
||||
@staticmethod
|
||||
def get_by_id(db: Session, logistics_id: int, invoice_id: int) -> Optional[models.InvoiceLogistics]:
|
||||
"""Get a logistics entry by ID"""
|
||||
return (
|
||||
db.query(models.InvoiceLogistics)
|
||||
.filter(
|
||||
models.InvoiceLogistics.logistics_id == logistics_id,
|
||||
models.InvoiceLogistics.invoice_id == invoice_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def get_all_by_invoice(db: Session, invoice_id: int) -> List[models.InvoiceLogistics]:
|
||||
"""Get all logistics entries for an invoice"""
|
||||
return (
|
||||
db.query(models.InvoiceLogistics)
|
||||
.filter(models.InvoiceLogistics.invoice_id == invoice_id)
|
||||
.all()
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def create(
|
||||
db: Session,
|
||||
logistics_data: schemas.InvoiceLogisticsCreate,
|
||||
invoice_id: int,
|
||||
tenant_id: int,
|
||||
company_id: int
|
||||
) -> models.InvoiceLogistics:
|
||||
"""Create a new logistics entry"""
|
||||
logistics_dict = logistics_data.model_dump()
|
||||
logistics_dict["invoice_id"] = invoice_id
|
||||
logistics_dict["tenant_id"] = tenant_id
|
||||
logistics_dict["company_id"] = company_id
|
||||
|
||||
new_logistics = models.InvoiceLogistics(**logistics_dict)
|
||||
db.add(new_logistics)
|
||||
db.commit()
|
||||
db.refresh(new_logistics)
|
||||
return new_logistics
|
||||
|
||||
@staticmethod
|
||||
def delete(db: Session, logistics_id: int, invoice_id: int) -> bool:
|
||||
"""Delete a logistics entry"""
|
||||
logistics = InvoiceLogisticsService.get_by_id(
|
||||
db, logistics_id, invoice_id)
|
||||
if logistics:
|
||||
db.delete(logistics)
|
||||
db.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
class InvoiceSalesDetailsService:
|
||||
"""Service for Invoice Sales Details operations"""
|
||||
|
||||
@staticmethod
|
||||
def get_by_id(db: Session, detail_id: int, invoice_id: int) -> Optional[models.InvoiceSalesDetails]:
|
||||
"""Get a sales detail entry by ID"""
|
||||
return (
|
||||
db.query(models.InvoiceSalesDetails)
|
||||
.filter(
|
||||
models.InvoiceSalesDetails.detail_id == detail_id,
|
||||
models.InvoiceSalesDetails.invoice_id == invoice_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def get_all_by_invoice(db: Session, invoice_id: int) -> List[models.InvoiceSalesDetails]:
|
||||
"""Get all sales details for an invoice"""
|
||||
return (
|
||||
db.query(models.InvoiceSalesDetails)
|
||||
.filter(models.InvoiceSalesDetails.invoice_id == invoice_id)
|
||||
.all()
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def create(
|
||||
db: Session,
|
||||
detail_data: schemas.InvoiceSalesDetailsCreate,
|
||||
invoice_id: int,
|
||||
tenant_id: int,
|
||||
company_id: int
|
||||
) -> models.InvoiceSalesDetails:
|
||||
"""Create a new sales detail entry"""
|
||||
detail_dict = detail_data.model_dump()
|
||||
detail_dict["invoice_id"] = invoice_id
|
||||
detail_dict["tenant_id"] = tenant_id
|
||||
detail_dict["company_id"] = company_id
|
||||
|
||||
new_detail = models.InvoiceSalesDetails(**detail_dict)
|
||||
db.add(new_detail)
|
||||
db.commit()
|
||||
db.refresh(new_detail)
|
||||
return new_detail
|
||||
|
||||
@staticmethod
|
||||
def delete(db: Session, detail_id: int, invoice_id: int) -> bool:
|
||||
"""Delete a sales detail entry"""
|
||||
detail = InvoiceSalesDetailsService.get_by_id(
|
||||
db, detail_id, invoice_id)
|
||||
if detail:
|
||||
db.delete(detail)
|
||||
db.commit()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
class InvoiceCollectionsService:
|
||||
"""Service for Invoice Collections operations"""
|
||||
|
||||
@staticmethod
|
||||
def get_by_id(db: Session, collection_id: int, invoice_id: int) -> Optional[models.InvoiceCollections]:
|
||||
"""Get a collection entry by ID"""
|
||||
return (
|
||||
db.query(models.InvoiceCollections)
|
||||
.filter(
|
||||
models.InvoiceCollections.collection_id == collection_id,
|
||||
models.InvoiceCollections.invoice_id == invoice_id,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def get_all_by_invoice(db: Session, invoice_id: int) -> List[models.InvoiceCollections]:
|
||||
"""Get all collections for an invoice"""
|
||||
return (
|
||||
db.query(models.InvoiceCollections)
|
||||
.filter(models.InvoiceCollections.invoice_id == invoice_id)
|
||||
.all()
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def create(
|
||||
db: Session,
|
||||
collection_data: schemas.InvoiceCollectionsCreate,
|
||||
invoice_id: int,
|
||||
tenant_id: int,
|
||||
company_id: int
|
||||
) -> models.InvoiceCollections:
|
||||
"""Create a new collection entry"""
|
||||
collection_dict = collection_data.model_dump()
|
||||
collection_dict["invoice_id"] = invoice_id
|
||||
collection_dict["tenant_id"] = tenant_id
|
||||
collection_dict["company_id"] = company_id
|
||||
|
||||
new_collection = models.InvoiceCollections(**collection_dict)
|
||||
db.add(new_collection)
|
||||
db.commit()
|
||||
db.refresh(new_collection)
|
||||
return new_collection
|
||||
|
||||
@staticmethod
|
||||
def delete(db: Session, collection_id: int, invoice_id: int) -> bool:
|
||||
"""Delete a collection entry"""
|
||||
collection = InvoiceCollectionsService.get_by_id(
|
||||
db, collection_id, invoice_id)
|
||||
if collection:
|
||||
db.delete(collection)
|
||||
db.commit()
|
||||
return True
|
||||
return False
|
||||
Reference in New Issue
Block a user