91 lines
3.1 KiB
Python
91 lines
3.1 KiB
Python
from typing import List, Optional
|
|
from sqlalchemy.orm import Session
|
|
from decimal import Decimal
|
|
|
|
from app.models.invoice import Invoice, InvoiceItem
|
|
from app.schemas.invoice import InvoiceCreate, InvoiceUpdate
|
|
|
|
|
|
class InvoiceService:
|
|
def get(self, db: Session, invoice_id: int) -> Optional[Invoice]:
|
|
return db.query(Invoice).filter(Invoice.id == invoice_id).first()
|
|
|
|
def get_by_number(self, db: Session, invoice_number: str) -> Optional[Invoice]:
|
|
return db.query(Invoice).filter(Invoice.invoice_number == invoice_number).first()
|
|
|
|
def get_multi_by_user(self, db: Session, user_id: int, skip: int = 0, limit: int = 100) -> List[Invoice]:
|
|
return db.query(Invoice).filter(Invoice.user_id == user_id).offset(skip).limit(limit).all()
|
|
|
|
def create(self, db: Session, obj_in: InvoiceCreate, user_id: int) -> Invoice:
|
|
db_obj = Invoice(
|
|
invoice_number=obj_in.invoice_number,
|
|
issue_date=obj_in.issue_date,
|
|
due_date=obj_in.due_date,
|
|
status=obj_in.status,
|
|
subtotal=obj_in.subtotal,
|
|
tax_rate=obj_in.tax_rate,
|
|
tax_amount=obj_in.tax_amount,
|
|
total_amount=obj_in.total_amount,
|
|
notes=obj_in.notes,
|
|
user_id=user_id,
|
|
customer_id=obj_in.customer_id,
|
|
)
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
|
|
for item in obj_in.items:
|
|
db_item = InvoiceItem(
|
|
description=item.description,
|
|
quantity=item.quantity,
|
|
unit_price=item.unit_price,
|
|
total_price=item.total_price,
|
|
invoice_id=db_obj.id,
|
|
)
|
|
db.add(db_item)
|
|
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def update(self, db: Session, db_obj: Invoice, obj_in: InvoiceUpdate) -> Invoice:
|
|
update_data = obj_in.dict(exclude_unset=True)
|
|
|
|
if "items" in update_data:
|
|
items_data = update_data.pop("items")
|
|
db.query(InvoiceItem).filter(InvoiceItem.invoice_id == db_obj.id).delete()
|
|
|
|
for item in items_data:
|
|
db_item = InvoiceItem(
|
|
description=item["description"],
|
|
quantity=item["quantity"],
|
|
unit_price=item["unit_price"],
|
|
total_price=item["total_price"],
|
|
invoice_id=db_obj.id,
|
|
)
|
|
db.add(db_item)
|
|
|
|
for field, value in update_data.items():
|
|
setattr(db_obj, field, value)
|
|
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def remove(self, db: Session, invoice_id: int) -> Invoice:
|
|
obj = db.query(Invoice).get(invoice_id)
|
|
db.delete(obj)
|
|
db.commit()
|
|
return obj
|
|
|
|
def calculate_totals(self, subtotal: Decimal, tax_rate: Decimal) -> dict:
|
|
tax_amount = subtotal * (tax_rate / 100)
|
|
total_amount = subtotal + tax_amount
|
|
return {
|
|
"tax_amount": tax_amount,
|
|
"total_amount": total_amount
|
|
}
|
|
|
|
|
|
invoice_service = InvoiceService() |