b51b13eb3e
- Set up FastAPI application with CORS support - Configure SQLite database connection - Create database models for users, clients, invoices, and line items - Set up Alembic for database migrations - Implement JWT-based authentication system - Create basic CRUD endpoints for users, clients, and invoices - Add PDF generation functionality - Implement activity logging - Update README with project information
230 lines
7.2 KiB
Python
230 lines
7.2 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from typing import List
|
|
from fastapi.responses import FileResponse
|
|
from sqlalchemy.orm import Session
|
|
from app.api.deps import get_db, get_current_user
|
|
from app.models.user import User
|
|
from app.schemas.invoice import Invoice, InvoiceCreate, InvoiceUpdate, InvoiceItem, InvoiceItemCreate
|
|
from app.crud import crud_invoice, crud_client
|
|
from app.core.logging import log_activity
|
|
from app.utils.pdf import generate_invoice_pdf
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/", response_model=List[Invoice])
|
|
async def get_invoices(
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Get all invoices for the current user.
|
|
"""
|
|
invoices = crud_invoice.get_multi_by_user(
|
|
db, user_id=current_user.id, skip=skip, limit=limit
|
|
)
|
|
|
|
# Calculate total for each invoice
|
|
for invoice in invoices:
|
|
total = sum(item.quantity * item.unit_price for item in invoice.items)
|
|
invoice.total = total
|
|
|
|
return invoices
|
|
|
|
|
|
@router.post("/", response_model=Invoice, status_code=status.HTTP_201_CREATED)
|
|
async def create_invoice(
|
|
invoice_in: InvoiceCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Create a new invoice.
|
|
"""
|
|
# Check if client exists and belongs to user
|
|
client = crud_client.get_by_id(db, client_id=invoice_in.client_id, user_id=current_user.id)
|
|
if not client:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Client not found",
|
|
)
|
|
|
|
invoice = crud_invoice.create(db, obj_in=invoice_in, user_id=current_user.id)
|
|
log_activity(current_user.id, "create", "invoice", invoice.id)
|
|
|
|
# Calculate total
|
|
total = sum(item.quantity * item.unit_price for item in invoice.items)
|
|
invoice.total = total
|
|
|
|
return invoice
|
|
|
|
|
|
@router.get("/{invoice_id}", response_model=Invoice)
|
|
async def get_invoice(
|
|
invoice_id: str,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Get an invoice by ID.
|
|
"""
|
|
invoice = crud_invoice.get_by_id(db, invoice_id=invoice_id, user_id=current_user.id)
|
|
if not invoice:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Invoice not found",
|
|
)
|
|
|
|
# Calculate total
|
|
total = sum(item.quantity * item.unit_price for item in invoice.items)
|
|
invoice.total = total
|
|
|
|
return invoice
|
|
|
|
|
|
@router.put("/{invoice_id}", response_model=Invoice)
|
|
async def update_invoice(
|
|
invoice_id: str,
|
|
invoice_in: InvoiceUpdate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Update an invoice.
|
|
"""
|
|
invoice = crud_invoice.get_by_id(db, invoice_id=invoice_id, user_id=current_user.id)
|
|
if not invoice:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Invoice not found",
|
|
)
|
|
|
|
# If client_id is changed, check if the new client exists and belongs to user
|
|
if invoice_in.client_id and invoice_in.client_id != invoice.client_id:
|
|
client = crud_client.get_by_id(db, client_id=invoice_in.client_id, user_id=current_user.id)
|
|
if not client:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Client not found",
|
|
)
|
|
|
|
invoice = crud_invoice.update(db, db_obj=invoice, obj_in=invoice_in)
|
|
log_activity(current_user.id, "update", "invoice", invoice.id)
|
|
|
|
# Calculate total
|
|
total = sum(item.quantity * item.unit_price for item in invoice.items)
|
|
invoice.total = total
|
|
|
|
return invoice
|
|
|
|
|
|
@router.delete("/{invoice_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None)
|
|
async def delete_invoice(
|
|
invoice_id: str,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Delete an invoice.
|
|
"""
|
|
invoice = crud_invoice.get_by_id(db, invoice_id=invoice_id, user_id=current_user.id)
|
|
if not invoice:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Invoice not found",
|
|
)
|
|
|
|
crud_invoice.remove(db, invoice_id=invoice_id, user_id=current_user.id)
|
|
log_activity(current_user.id, "delete", "invoice", invoice_id)
|
|
return None
|
|
|
|
|
|
@router.post("/{invoice_id}/items", response_model=InvoiceItem, status_code=status.HTTP_201_CREATED)
|
|
async def add_invoice_item(
|
|
invoice_id: str,
|
|
item_in: InvoiceItemCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Add an item to an invoice.
|
|
"""
|
|
invoice = crud_invoice.get_by_id(db, invoice_id=invoice_id, user_id=current_user.id)
|
|
if not invoice:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Invoice not found",
|
|
)
|
|
|
|
item = crud_invoice.add_item(db, invoice_id=invoice_id, user_id=current_user.id, obj_in=item_in)
|
|
log_activity(current_user.id, "create", "invoice item", item.id)
|
|
return item
|
|
|
|
|
|
@router.delete("/{invoice_id}/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None)
|
|
async def delete_invoice_item(
|
|
invoice_id: str,
|
|
item_id: str,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Delete an item from an invoice.
|
|
"""
|
|
invoice = crud_invoice.get_by_id(db, invoice_id=invoice_id, user_id=current_user.id)
|
|
if not invoice:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Invoice not found",
|
|
)
|
|
|
|
item = crud_invoice.remove_item(db, item_id=item_id, invoice_id=invoice_id, user_id=current_user.id)
|
|
if not item:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Invoice item not found",
|
|
)
|
|
|
|
log_activity(current_user.id, "delete", "invoice item", item_id)
|
|
return None
|
|
|
|
|
|
@router.get("/{invoice_id}/pdf")
|
|
async def get_invoice_pdf(
|
|
invoice_id: str,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Generate and download a PDF for an invoice.
|
|
"""
|
|
invoice = crud_invoice.get_by_id(db, invoice_id=invoice_id, user_id=current_user.id)
|
|
if not invoice:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Invoice not found",
|
|
)
|
|
|
|
client = crud_client.get_by_id(db, client_id=invoice.client_id, user_id=current_user.id)
|
|
if not client:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Client not found",
|
|
)
|
|
|
|
try:
|
|
pdf_path = generate_invoice_pdf(invoice, client)
|
|
log_activity(current_user.id, "generate", "invoice PDF", invoice_id)
|
|
|
|
return FileResponse(
|
|
path=pdf_path,
|
|
filename=f"invoice_{invoice.invoice_number}.pdf",
|
|
media_type="application/pdf",
|
|
)
|
|
except Exception as e:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=f"Error generating PDF: {str(e)}",
|
|
) |