bad5cc0eba
Features: - JWT authentication with user registration and login - Customer management with full CRUD operations - Invoice management with automatic calculations - Multi-tenant data isolation by user - SQLite database with Alembic migrations - RESTful API with comprehensive documentation - Tax calculations and invoice status tracking - Code formatted with Ruff linting
143 lines
4.5 KiB
Python
143 lines
4.5 KiB
Python
from sqlalchemy.orm import Session
|
|
from app.models.invoice import Invoice, InvoiceItem, InvoiceStatus
|
|
from app.schemas.invoice import InvoiceCreate, InvoiceUpdate
|
|
from typing import List, Optional
|
|
from decimal import Decimal
|
|
import uuid
|
|
|
|
|
|
class InvoiceService:
|
|
def __init__(self, db: Session):
|
|
self.db = db
|
|
|
|
def get_invoices(
|
|
self, user_id: int, skip: int = 0, limit: int = 100
|
|
) -> List[Invoice]:
|
|
return (
|
|
self.db.query(Invoice)
|
|
.filter(Invoice.user_id == user_id)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def get_invoice(self, invoice_id: int, user_id: int) -> Optional[Invoice]:
|
|
return (
|
|
self.db.query(Invoice)
|
|
.filter(Invoice.id == invoice_id, Invoice.user_id == user_id)
|
|
.first()
|
|
)
|
|
|
|
def create_invoice(self, invoice_data: InvoiceCreate, user_id: int) -> Invoice:
|
|
# Generate unique invoice number
|
|
invoice_number = f"INV-{uuid.uuid4().hex[:8].upper()}"
|
|
|
|
# Calculate totals
|
|
subtotal = sum(item.quantity * item.unit_price for item in invoice_data.items)
|
|
tax_amount = (
|
|
subtotal * (invoice_data.tax_rate / 100)
|
|
if invoice_data.tax_rate
|
|
else Decimal("0")
|
|
)
|
|
total = subtotal + tax_amount
|
|
|
|
# Create invoice
|
|
invoice = Invoice(
|
|
invoice_number=invoice_number,
|
|
customer_id=invoice_data.customer_id,
|
|
user_id=user_id,
|
|
due_date=invoice_data.due_date,
|
|
tax_rate=invoice_data.tax_rate or Decimal("0"),
|
|
subtotal=subtotal,
|
|
tax_amount=tax_amount,
|
|
total=total,
|
|
notes=invoice_data.notes,
|
|
)
|
|
|
|
self.db.add(invoice)
|
|
self.db.flush() # Get the invoice ID
|
|
|
|
# Create invoice items
|
|
for item_data in invoice_data.items:
|
|
item_total = item_data.quantity * item_data.unit_price
|
|
item = InvoiceItem(
|
|
invoice_id=invoice.id,
|
|
description=item_data.description,
|
|
quantity=item_data.quantity,
|
|
unit_price=item_data.unit_price,
|
|
total_price=item_total,
|
|
)
|
|
self.db.add(item)
|
|
|
|
self.db.commit()
|
|
self.db.refresh(invoice)
|
|
return invoice
|
|
|
|
def update_invoice(
|
|
self, invoice_id: int, invoice_data: InvoiceUpdate, user_id: int
|
|
) -> Optional[Invoice]:
|
|
invoice = self.get_invoice(invoice_id, user_id)
|
|
if not invoice:
|
|
return None
|
|
|
|
# Update basic fields
|
|
update_data = invoice_data.dict(exclude_unset=True, exclude={"items"})
|
|
for field, value in update_data.items():
|
|
setattr(invoice, field, value)
|
|
|
|
# Update items if provided
|
|
if invoice_data.items is not None:
|
|
# Delete existing items
|
|
self.db.query(InvoiceItem).filter(
|
|
InvoiceItem.invoice_id == invoice_id
|
|
).delete()
|
|
|
|
# Add new items
|
|
subtotal = Decimal("0")
|
|
for item_data in invoice_data.items:
|
|
item_total = item_data.quantity * item_data.unit_price
|
|
subtotal += item_total
|
|
item = InvoiceItem(
|
|
invoice_id=invoice.id,
|
|
description=item_data.description,
|
|
quantity=item_data.quantity,
|
|
unit_price=item_data.unit_price,
|
|
total_price=item_total,
|
|
)
|
|
self.db.add(item)
|
|
|
|
# Recalculate totals
|
|
tax_amount = (
|
|
subtotal * (invoice.tax_rate / 100)
|
|
if invoice.tax_rate
|
|
else Decimal("0")
|
|
)
|
|
invoice.subtotal = subtotal
|
|
invoice.tax_amount = tax_amount
|
|
invoice.total = subtotal + tax_amount
|
|
|
|
self.db.commit()
|
|
self.db.refresh(invoice)
|
|
return invoice
|
|
|
|
def delete_invoice(self, invoice_id: int, user_id: int) -> bool:
|
|
invoice = self.get_invoice(invoice_id, user_id)
|
|
if not invoice:
|
|
return False
|
|
|
|
self.db.delete(invoice)
|
|
self.db.commit()
|
|
return True
|
|
|
|
def update_invoice_status(
|
|
self, invoice_id: int, status: InvoiceStatus, user_id: int
|
|
) -> Optional[Invoice]:
|
|
invoice = self.get_invoice(invoice_id, user_id)
|
|
if not invoice:
|
|
return None
|
|
|
|
invoice.status = status
|
|
self.db.commit()
|
|
self.db.refresh(invoice)
|
|
return invoice
|