sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
11ebb2116a56cdbcbbc3af59541fe77e50fcd2a1
orion/app/services/invoice_service.py
Samir Boulahtit 319fba5d39 fix: resolve architecture validation violations 
- Add invoice exceptions module with proper exception hierarchy
- Replace HTTPException with service-layer exceptions in invoice API
- Add InvoicePDFGeneratedResponse and InvoiceStatsResponse Pydantic models
- Replace db.commit() with db.flush() in services for proper transaction control
- Update invoice service to use exceptions from app/exceptions/invoice.py

All 14 errors and 14 warnings are now resolved. Validation passes with
only INFO-level findings.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-24 18:23:49 +01:00

661 lines
21 KiB
Python
Raw Blame History

# app/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)
VAT Logic:
- Luxembourg domestic: 17% (standard), 8% (reduced), 3% (super-reduced), 14% (intermediate)
- EU cross-border B2C with OSS: Use destination country VAT rate
- EU cross-border B2C without OSS: Use Luxembourg VAT rate (origin principle)
- EU B2B with valid VAT number: Reverse charge (0% VAT)
"""
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,
InvoicePDFGenerationException,
InvoicePDFNotFoundException,
InvoiceSettingsAlreadyExistException,
InvoiceSettingsNotFoundException,
InvoiceValidationException,
OrderNotFoundException,
)
from models.database.invoice import (
Invoice,
InvoiceStatus,
VATRegime,
VendorInvoiceSettings,
)
from models.database.order import Order
from models.database.vendor import Vendor
from models.schema.invoice import (
InvoiceBuyerDetails,
InvoiceCreate,
InvoiceLineItem,
InvoiceManualCreate,
InvoiceSellerDetails,
VendorInvoiceSettingsCreate,
VendorInvoiceSettingsUpdate,
)
logger = logging.getLogger(__name__)
# EU VAT rates by country code (2024 standard rates)
EU_VAT_RATES: dict[str, Decimal] = {
"AT": Decimal("20.00"), # Austria
"BE": Decimal("21.00"), # Belgium
"BG": Decimal("20.00"), # Bulgaria
"HR": Decimal("25.00"), # Croatia
"CY": Decimal("19.00"), # Cyprus
"CZ": Decimal("21.00"), # Czech Republic
"DK": Decimal("25.00"), # Denmark
"EE": Decimal("22.00"), # Estonia
"FI": Decimal("24.00"), # Finland
"FR": Decimal("20.00"), # France
"DE": Decimal("19.00"), # Germany
"GR": Decimal("24.00"), # Greece
"HU": Decimal("27.00"), # Hungary
"IE": Decimal("23.00"), # Ireland
"IT": Decimal("22.00"), # Italy
"LV": Decimal("21.00"), # Latvia
"LT": Decimal("21.00"), # Lithuania
"LU": Decimal("17.00"), # Luxembourg (standard)
"MT": Decimal("18.00"), # Malta
"NL": Decimal("21.00"), # Netherlands
"PL": Decimal("23.00"), # Poland
"PT": Decimal("23.00"), # Portugal
"RO": Decimal("19.00"), # Romania
"SK": Decimal("20.00"), # Slovakia
"SI": Decimal("22.00"), # Slovenia
"ES": Decimal("21.00"), # Spain
"SE": Decimal("25.00"), # Sweden
}
# Luxembourg specific VAT rates
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.
Returns: (regime, vat_rate, destination_country)
"""
seller_country = seller_country.upper()
buyer_country = buyer_country.upper()
# Same country = domestic VAT
if seller_country == buyer_country:
vat_rate = self.get_vat_rate_for_country(seller_country)
return VATRegime.DOMESTIC, vat_rate, None
# Different EU countries
if buyer_country in EU_VAT_RATES:
# B2B with valid VAT number = reverse charge
if buyer_vat_number:
return VATRegime.REVERSE_CHARGE, Decimal("0.00"), buyer_country
# B2C cross-border
if seller_oss_registered:
# OSS: use destination country VAT
vat_rate = self.get_vat_rate_for_country(buyer_country)
return VATRegime.OSS, vat_rate, buyer_country
else:
# No OSS: use origin country VAT
vat_rate = self.get_vat_rate_for_country(seller_country)
return VATRegime.ORIGIN, vat_rate, buyer_country
# Non-EU = VAT exempt (export)
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."""
# Check if settings already exist
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.
Used for initial setup based on existing vendor data.
"""
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, # Would need to parse from address
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}"
# Increment counter
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.
Captures snapshots of seller/buyer details and calculates VAT.
"""
# Get invoice settings
settings = self.get_settings_or_raise(db, vendor_id)
# Get order
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")
# Check for existing invoice
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}")
# Determine VAT regime
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, # TODO: Add B2B VAT number support
seller_oss_registered=settings.is_oss_registered,
)
# Build seller details snapshot
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,
}
# Build buyer details snapshot
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, # TODO: B2B support
}
if order.bill_company:
buyer_details["company"] = order.bill_company
# Build line items from order items
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,
})
# Calculate amounts
subtotal_cents = sum(item["total_cents"] for item in line_items)
# Calculate VAT
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
# Get VAT label
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)
# Generate invoice number
invoice_number = self._get_next_invoice_number(db, settings)
# Create invoice
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 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.
Returns: (invoices, total_count)
"""
query = db.query(Invoice).filter(Invoice.vendor_id == vendor_id)
if status:
query = query.filter(Invoice.status == status)
# Get total count
total = query.count()
# Apply pagination and order
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)
# Validate status transition
valid_statuses = [s.value for s in InvoiceStatus]
if new_status not in valid_statuses:
raise ValidationException(f"Invalid status: {new_status}")
# Cannot change cancelled invoices
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.
Returns path to the generated PDF.
"""
from app.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.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