sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
de83875d0ad29162ef11ff4e6a21139fdb55cc51
orion/app/modules/orders/services/invoice_service.py
Samir Boulahtit de83875d0a refactor: migrate modules from re-exports to canonical implementations 
Move actual code implementations into module directories:
- orders: 5 services, 4 models, order/invoice schemas
- inventory: 3 services, 2 models, 30+ schemas
- customers: 3 services, 2 models, customer schemas
- messaging: 3 services, 2 models, message/notification schemas
- monitoring: background_tasks_service
- marketplace: 5+ services including letzshop submodule
- dev_tools: code_quality_service, test_runner_service
- billing: billing_service
- contracts: definition.py

Legacy files in app/services/, models/database/, models/schema/
now re-export from canonical module locations for backwards
compatibility. Architecture validator passes with 0 errors.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-29 21:28:56 +01:00

588 lines
19 KiB
Python
Raw Blame History

# app/modules/orders/services/invoice_service.py
"""
Invoice service for generating and managing invoices.
Handles:
- Vendor invoice settings management
- Invoice generation from orders
- VAT calculation (Luxembourg, EU, B2B reverse charge)
- Invoice number sequencing
- PDF generation (via separate module)
"""
import logging
from datetime import UTC, datetime
from decimal import Decimal
from typing import Any
from sqlalchemy import and_, func
from sqlalchemy.orm import Session
from app.exceptions import ValidationException
from app.exceptions.invoice import (
InvoiceNotFoundException,
InvoiceSettingsNotFoundException,
OrderNotFoundException,
)
from app.modules.orders.models.invoice import (
Invoice,
InvoiceStatus,
VATRegime,
VendorInvoiceSettings,
)
from app.modules.orders.models.order import Order
from app.modules.orders.schemas.invoice import (
VendorInvoiceSettingsCreate,
VendorInvoiceSettingsUpdate,
)
from models.database.vendor import Vendor
logger = logging.getLogger(__name__)
# EU VAT rates by country code (2024 standard rates)
EU_VAT_RATES: dict[str, Decimal] = {
"AT": Decimal("20.00"),
"BE": Decimal("21.00"),
"BG": Decimal("20.00"),
"HR": Decimal("25.00"),
"CY": Decimal("19.00"),
"CZ": Decimal("21.00"),
"DK": Decimal("25.00"),
"EE": Decimal("22.00"),
"FI": Decimal("24.00"),
"FR": Decimal("20.00"),
"DE": Decimal("19.00"),
"GR": Decimal("24.00"),
"HU": Decimal("27.00"),
"IE": Decimal("23.00"),
"IT": Decimal("22.00"),
"LV": Decimal("21.00"),
"LT": Decimal("21.00"),
"LU": Decimal("17.00"),
"MT": Decimal("18.00"),
"NL": Decimal("21.00"),
"PL": Decimal("23.00"),
"PT": Decimal("23.00"),
"RO": Decimal("19.00"),
"SK": Decimal("20.00"),
"SI": Decimal("22.00"),
"ES": Decimal("21.00"),
"SE": Decimal("25.00"),
}
LU_VAT_RATES = {
"standard": Decimal("17.00"),
"intermediate": Decimal("14.00"),
"reduced": Decimal("8.00"),
"super_reduced": Decimal("3.00"),
}
class InvoiceService:
"""Service for invoice operations."""
# =========================================================================
# VAT Calculation
# =========================================================================
def get_vat_rate_for_country(self, country_iso: str) -> Decimal:
"""Get standard VAT rate for EU country."""
return EU_VAT_RATES.get(country_iso.upper(), Decimal("0.00"))
def get_vat_rate_label(self, country_iso: str, vat_rate: Decimal) -> str:
"""Get human-readable VAT rate label."""
country_names = {
"AT": "Austria", "BE": "Belgium", "BG": "Bulgaria", "HR": "Croatia",
"CY": "Cyprus", "CZ": "Czech Republic", "DK": "Denmark", "EE": "Estonia",
"FI": "Finland", "FR": "France", "DE": "Germany", "GR": "Greece",
"HU": "Hungary", "IE": "Ireland", "IT": "Italy", "LV": "Latvia",
"LT": "Lithuania", "LU": "Luxembourg", "MT": "Malta", "NL": "Netherlands",
"PL": "Poland", "PT": "Portugal", "RO": "Romania", "SK": "Slovakia",
"SI": "Slovenia", "ES": "Spain", "SE": "Sweden",
}
country_name = country_names.get(country_iso.upper(), country_iso)
return f"{country_name} VAT {vat_rate}%"
def determine_vat_regime(
self,
seller_country: str,
buyer_country: str,
buyer_vat_number: str | None,
seller_oss_registered: bool,
) -> tuple[VATRegime, Decimal, str | None]:
"""Determine VAT regime and rate for invoice."""
seller_country = seller_country.upper()
buyer_country = buyer_country.upper()
if seller_country == buyer_country:
vat_rate = self.get_vat_rate_for_country(seller_country)
return VATRegime.DOMESTIC, vat_rate, None
if buyer_country in EU_VAT_RATES:
if buyer_vat_number:
return VATRegime.REVERSE_CHARGE, Decimal("0.00"), buyer_country
if seller_oss_registered:
vat_rate = self.get_vat_rate_for_country(buyer_country)
return VATRegime.OSS, vat_rate, buyer_country
else:
vat_rate = self.get_vat_rate_for_country(seller_country)
return VATRegime.ORIGIN, vat_rate, buyer_country
return VATRegime.EXEMPT, Decimal("0.00"), buyer_country
# =========================================================================
# Invoice Settings Management
# =========================================================================
def get_settings(
self, db: Session, vendor_id: int
) -> VendorInvoiceSettings | None:
"""Get vendor invoice settings."""
return (
db.query(VendorInvoiceSettings)
.filter(VendorInvoiceSettings.vendor_id == vendor_id)
.first()
)
def get_settings_or_raise(
self, db: Session, vendor_id: int
) -> VendorInvoiceSettings:
"""Get vendor invoice settings or raise exception."""
settings = self.get_settings(db, vendor_id)
if not settings:
raise InvoiceSettingsNotFoundException(vendor_id)
return settings
def create_settings(
self,
db: Session,
vendor_id: int,
data: VendorInvoiceSettingsCreate,
) -> VendorInvoiceSettings:
"""Create vendor invoice settings."""
existing = self.get_settings(db, vendor_id)
if existing:
raise ValidationException(
"Invoice settings already exist for this vendor"
)
settings = VendorInvoiceSettings(
vendor_id=vendor_id,
**data.model_dump(),
)
db.add(settings)
db.flush()
db.refresh(settings)
logger.info(f"Created invoice settings for vendor {vendor_id}")
return settings
def update_settings(
self,
db: Session,
vendor_id: int,
data: VendorInvoiceSettingsUpdate,
) -> VendorInvoiceSettings:
"""Update vendor invoice settings."""
settings = self.get_settings_or_raise(db, vendor_id)
update_data = data.model_dump(exclude_unset=True)
for key, value in update_data.items():
setattr(settings, key, value)
settings.updated_at = datetime.now(UTC)
db.flush()
db.refresh(settings)
logger.info(f"Updated invoice settings for vendor {vendor_id}")
return settings
def create_settings_from_vendor(
self,
db: Session,
vendor: Vendor,
) -> VendorInvoiceSettings:
"""Create invoice settings from vendor/company info."""
company = vendor.company
settings = VendorInvoiceSettings(
vendor_id=vendor.id,
company_name=company.legal_name if company else vendor.name,
company_address=vendor.effective_business_address,
company_city=None,
company_postal_code=None,
company_country="LU",
vat_number=vendor.effective_tax_number,
is_vat_registered=bool(vendor.effective_tax_number),
)
db.add(settings)
db.flush()
db.refresh(settings)
logger.info(f"Created invoice settings from vendor data for vendor {vendor.id}")
return settings
# =========================================================================
# Invoice Number Generation
# =========================================================================
def _get_next_invoice_number(
self, db: Session, settings: VendorInvoiceSettings
) -> str:
"""Generate next invoice number and increment counter."""
number = str(settings.invoice_next_number).zfill(settings.invoice_number_padding)
invoice_number = f"{settings.invoice_prefix}{number}"
settings.invoice_next_number += 1
db.flush()
return invoice_number
# =========================================================================
# Invoice Creation
# =========================================================================
def create_invoice_from_order(
self,
db: Session,
vendor_id: int,
order_id: int,
notes: str | None = None,
) -> Invoice:
"""Create an invoice from an order."""
settings = self.get_settings_or_raise(db, vendor_id)
order = (
db.query(Order)
.filter(and_(Order.id == order_id, Order.vendor_id == vendor_id))
.first()
)
if not order:
raise OrderNotFoundException(f"Order {order_id} not found")
existing = (
db.query(Invoice)
.filter(and_(Invoice.order_id == order_id, Invoice.vendor_id == vendor_id))
.first()
)
if existing:
raise ValidationException(f"Invoice already exists for order {order_id}")
buyer_country = order.bill_country_iso
vat_regime, vat_rate, destination_country = self.determine_vat_regime(
seller_country=settings.company_country,
buyer_country=buyer_country,
buyer_vat_number=None,
seller_oss_registered=settings.is_oss_registered,
)
seller_details = {
"company_name": settings.company_name,
"address": settings.company_address,
"city": settings.company_city,
"postal_code": settings.company_postal_code,
"country": settings.company_country,
"vat_number": settings.vat_number,
}
buyer_details = {
"name": f"{order.bill_first_name} {order.bill_last_name}".strip(),
"email": order.customer_email,
"address": order.bill_address_line_1,
"city": order.bill_city,
"postal_code": order.bill_postal_code,
"country": order.bill_country_iso,
"vat_number": None,
}
if order.bill_company:
buyer_details["company"] = order.bill_company
line_items = []
for item in order.items:
line_items.append({
"description": item.product_name,
"quantity": item.quantity,
"unit_price_cents": item.unit_price_cents,
"total_cents": item.total_price_cents,
"sku": item.product_sku,
"ean": item.gtin,
})
subtotal_cents = sum(item["total_cents"] for item in line_items)
if vat_rate > 0:
vat_amount_cents = int(subtotal_cents * float(vat_rate) / 100)
else:
vat_amount_cents = 0
total_cents = subtotal_cents + vat_amount_cents
vat_rate_label = None
if vat_rate > 0:
if destination_country:
vat_rate_label = self.get_vat_rate_label(destination_country, vat_rate)
else:
vat_rate_label = self.get_vat_rate_label(settings.company_country, vat_rate)
invoice_number = self._get_next_invoice_number(db, settings)
invoice = Invoice(
vendor_id=vendor_id,
order_id=order_id,
invoice_number=invoice_number,
invoice_date=datetime.now(UTC),
status=InvoiceStatus.DRAFT.value,
seller_details=seller_details,
buyer_details=buyer_details,
line_items=line_items,
vat_regime=vat_regime.value,
destination_country=destination_country,
vat_rate=vat_rate,
vat_rate_label=vat_rate_label,
currency=order.currency,
subtotal_cents=subtotal_cents,
vat_amount_cents=vat_amount_cents,
total_cents=total_cents,
payment_terms=settings.payment_terms,
bank_details={
"bank_name": settings.bank_name,
"iban": settings.bank_iban,
"bic": settings.bank_bic,
} if settings.bank_iban else None,
footer_text=settings.footer_text,
notes=notes,
)
db.add(invoice)
db.flush()
db.refresh(invoice)
logger.info(
f"Created invoice {invoice_number} for order {order_id} "
f"(vendor={vendor_id}, total={total_cents/100:.2f} EUR, VAT={vat_regime.value})"
)
return invoice
# =========================================================================
# Invoice Retrieval
# =========================================================================
def get_invoice(
self, db: Session, vendor_id: int, invoice_id: int
) -> Invoice | None:
"""Get invoice by ID."""
return (
db.query(Invoice)
.filter(and_(Invoice.id == invoice_id, Invoice.vendor_id == vendor_id))
.first()
)
def get_invoice_or_raise(
self, db: Session, vendor_id: int, invoice_id: int
) -> Invoice:
"""Get invoice by ID or raise exception."""
invoice = self.get_invoice(db, vendor_id, invoice_id)
if not invoice:
raise InvoiceNotFoundException(invoice_id)
return invoice
def get_invoice_by_number(
self, db: Session, vendor_id: int, invoice_number: str
) -> Invoice | None:
"""Get invoice by invoice number."""
return (
db.query(Invoice)
.filter(
and_(
Invoice.invoice_number == invoice_number,
Invoice.vendor_id == vendor_id,
)
)
.first()
)
def get_invoice_by_order_id(
self, db: Session, vendor_id: int, order_id: int
) -> Invoice | None:
"""Get invoice by order ID."""
return (
db.query(Invoice)
.filter(
and_(
Invoice.order_id == order_id,
Invoice.vendor_id == vendor_id,
)
)
.first()
)
def list_invoices(
self,
db: Session,
vendor_id: int,
status: str | None = None,
page: int = 1,
per_page: int = 20,
) -> tuple[list[Invoice], int]:
"""List invoices for vendor with pagination."""
query = db.query(Invoice).filter(Invoice.vendor_id == vendor_id)
if status:
query = query.filter(Invoice.status == status)
total = query.count()
invoices = (
query.order_by(Invoice.invoice_date.desc())
.offset((page - 1) * per_page)
.limit(per_page)
.all()
)
return invoices, total
# =========================================================================
# Invoice Status Management
# =========================================================================
def update_status(
self,
db: Session,
vendor_id: int,
invoice_id: int,
new_status: str,
) -> Invoice:
"""Update invoice status."""
invoice = self.get_invoice_or_raise(db, vendor_id, invoice_id)
valid_statuses = [s.value for s in InvoiceStatus]
if new_status not in valid_statuses:
raise ValidationException(f"Invalid status: {new_status}")
if invoice.status == InvoiceStatus.CANCELLED.value:
raise ValidationException("Cannot change status of cancelled invoice")
invoice.status = new_status
invoice.updated_at = datetime.now(UTC)
db.flush()
db.refresh(invoice)
logger.info(f"Updated invoice {invoice.invoice_number} status to {new_status}")
return invoice
def mark_as_issued(
self, db: Session, vendor_id: int, invoice_id: int
) -> Invoice:
"""Mark invoice as issued."""
return self.update_status(db, vendor_id, invoice_id, InvoiceStatus.ISSUED.value)
def mark_as_paid(
self, db: Session, vendor_id: int, invoice_id: int
) -> Invoice:
"""Mark invoice as paid."""
return self.update_status(db, vendor_id, invoice_id, InvoiceStatus.PAID.value)
def cancel_invoice(
self, db: Session, vendor_id: int, invoice_id: int
) -> Invoice:
"""Cancel invoice."""
return self.update_status(db, vendor_id, invoice_id, InvoiceStatus.CANCELLED.value)
# =========================================================================
# Statistics
# =========================================================================
def get_invoice_stats(
self, db: Session, vendor_id: int
) -> dict[str, Any]:
"""Get invoice statistics for vendor."""
total_count = (
db.query(func.count(Invoice.id))
.filter(Invoice.vendor_id == vendor_id)
.scalar()
or 0
)
total_revenue = (
db.query(func.sum(Invoice.total_cents))
.filter(
and_(
Invoice.vendor_id == vendor_id,
Invoice.status.in_([
InvoiceStatus.ISSUED.value,
InvoiceStatus.PAID.value,
]),
)
)
.scalar()
or 0
)
draft_count = (
db.query(func.count(Invoice.id))
.filter(
and_(
Invoice.vendor_id == vendor_id,
Invoice.status == InvoiceStatus.DRAFT.value,
)
)
.scalar()
or 0
)
paid_count = (
db.query(func.count(Invoice.id))
.filter(
and_(
Invoice.vendor_id == vendor_id,
Invoice.status == InvoiceStatus.PAID.value,
)
)
.scalar()
or 0
)
return {
"total_invoices": total_count,
"total_revenue_cents": total_revenue,
"total_revenue": total_revenue / 100 if total_revenue else 0,
"draft_count": draft_count,
"paid_count": paid_count,
}
# =========================================================================
# PDF Generation
# =========================================================================
def generate_pdf(
self,
db: Session,
vendor_id: int,
invoice_id: int,
force_regenerate: bool = False,
) -> str:
"""Generate PDF for an invoice."""
from app.modules.orders.services.invoice_pdf_service import invoice_pdf_service
invoice = self.get_invoice_or_raise(db, vendor_id, invoice_id)
return invoice_pdf_service.generate_pdf(db, invoice, force_regenerate)
def get_pdf_path(
self,
db: Session,
vendor_id: int,
invoice_id: int,
) -> str | None:
"""Get PDF path for an invoice if it exists."""
from app.modules.orders.services.invoice_pdf_service import invoice_pdf_service
invoice = self.get_invoice_or_raise(db, vendor_id, invoice_id)
return invoice_pdf_service.get_pdf_path(invoice)
# Singleton instance
invoice_service = InvoiceService()
Reference in New Issue View Git Blame Copy Permalink