from typing import List, Optional, Union, Dict, Any from sqlalchemy.orm import Session, joinedload from app.models.invoice import Invoice, InvoiceItem from app.schemas.invoice import InvoiceCreate, InvoiceUpdate, InvoiceItemCreate def get_invoice(db: Session, invoice_id: int) -> Optional[Invoice]: """ Get invoice by ID with items """ return ( db.query(Invoice) .options(joinedload(Invoice.items)) .filter(Invoice.id == invoice_id) .first() ) def get_invoices_by_user( db: Session, user_id: int, skip: int = 0, limit: int = 100 ) -> List[Invoice]: """ Get all invoices for a user """ return ( db.query(Invoice) .filter(Invoice.user_id == user_id) .order_by(Invoice.created_at.desc()) .offset(skip) .limit(limit) .all() ) def get_invoices_by_client( db: Session, client_id: int, user_id: int, skip: int = 0, limit: int = 100 ) -> List[Invoice]: """ Get all invoices for a client """ return ( db.query(Invoice) .filter(Invoice.client_id == client_id, Invoice.user_id == user_id) .order_by(Invoice.created_at.desc()) .offset(skip) .limit(limit) .all() ) def create_invoice(db: Session, obj_in: InvoiceCreate, user_id: int) -> Invoice: """ Create new invoice with items """ # Create invoice db_obj = Invoice( user_id=user_id, client_id=obj_in.client_id, invoice_number=obj_in.invoice_number, status=obj_in.status, issued_date=obj_in.issued_date, due_date=obj_in.due_date, notes=obj_in.notes, total_amount=0.0, # Will be calculated after items are added ) db.add(db_obj) db.commit() db.refresh(db_obj) # Create invoice items for item_in in obj_in.items: create_invoice_item(db, item_in, db_obj.id) # Update total amount total_amount = calculate_invoice_total(db, db_obj.id) db_obj.total_amount = total_amount db.add(db_obj) db.commit() db.refresh(db_obj) return db_obj def update_invoice( db: Session, *, db_obj: Invoice, obj_in: Union[InvoiceUpdate, Dict[str, Any]] ) -> Invoice: """ Update invoice """ if isinstance(obj_in, dict): update_data = obj_in else: update_data = obj_in.dict(exclude_unset=True) # Handle items separately items = update_data.pop("items", None) # Update invoice fields for field in update_data: if hasattr(db_obj, field): setattr(db_obj, field, update_data[field]) db.add(db_obj) db.commit() db.refresh(db_obj) # Update items if provided if items: # Delete existing items db.query(InvoiceItem).filter(InvoiceItem.invoice_id == db_obj.id).delete() db.commit() # Create new items for item_in in items: create_invoice_item(db, item_in, db_obj.id) # Update total amount total_amount = calculate_invoice_total(db, db_obj.id) db_obj.total_amount = total_amount db.add(db_obj) db.commit() db.refresh(db_obj) return db_obj def delete_invoice(db: Session, *, invoice_id: int) -> Invoice: """ Delete invoice """ invoice = db.query(Invoice).filter(Invoice.id == invoice_id).first() if invoice: # Delete all related items first db.query(InvoiceItem).filter(InvoiceItem.invoice_id == invoice_id).delete() db.commit() # Delete the invoice db.delete(invoice) db.commit() return invoice def create_invoice_item(db: Session, obj_in: InvoiceItemCreate, invoice_id: int) -> InvoiceItem: """ Create new invoice item """ db_obj = InvoiceItem( invoice_id=invoice_id, description=obj_in.description, quantity=obj_in.quantity, unit_price=obj_in.unit_price, ) db.add(db_obj) db.commit() db.refresh(db_obj) return db_obj def calculate_invoice_total(db: Session, invoice_id: int) -> float: """ Calculate the total amount for an invoice """ items = db.query(InvoiceItem).filter(InvoiceItem.invoice_id == invoice_id).all() total = sum(item.quantity * item.unit_price for item in items) return total def update_invoice_pdf_path(db: Session, invoice_id: int, pdf_path: str) -> Invoice: """ Update the PDF path for an invoice """ invoice = db.query(Invoice).filter(Invoice.id == invoice_id).first() if invoice: invoice.pdf_path = pdf_path db.add(invoice) db.commit() db.refresh(invoice) return invoice