import csv
import io
from datetime import date
from fastapi import APIRouter, Depends, Query
from fastapi.responses import StreamingResponse
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from app.auth.dependencies import get_current_user
from app.database import get_session
from app.models.transaction import Transaction
from app.models.user import User
from app.utils import utcnow
router = APIRouter(prefix="/export", tags=["export"])
def _month_bounds(month: str) -> tuple[date, date]:
year, mon = map(int, month.split("-"))
start = date(year, mon, 1)
if mon == 12:
end = date(year + 1, 1, 1)
else:
end = date(year, mon + 1, 1)
return start, end
async def _fetch_transactions(
session: AsyncSession,
user_id,
month: str,
) -> list[Transaction]:
start, end = _month_bounds(month)
result = await session.execute(
select(Transaction)
.options(selectinload(Transaction.category))
.where(
Transaction.user_id == user_id,
Transaction.deleted_at.is_(None),
Transaction.transaction_date >= start,
Transaction.transaction_date < end,
)
.order_by(Transaction.transaction_date, Transaction.created_at)
)
return list(result.scalars().all())
@router.get("/csv")
async def export_csv(
month: str = Query(default=None, description="Month YYYY-MM"),
session: AsyncSession = Depends(get_session),
current_user: User = Depends(get_current_user),
) -> StreamingResponse:
if month is None:
now = utcnow()
month = f"{now.year}-{now.month:02d}"
transactions = await _fetch_transactions(session, current_user.id, month)
output = io.StringIO()
writer = csv.writer(output, delimiter=";")
writer.writerow(["Date", "Type", "Catégorie", "Description", "Montant (€)"])
for tx in transactions:
amount_eur = tx.amount_cents / 100
if tx.type.value == "expense":
amount_eur = -amount_eur
writer.writerow([
tx.transaction_date.isoformat(),
tx.type.value,
tx.category.name,
tx.description or "",
f"{amount_eur:.2f}",
])
output.seek(0)
filename = f"transactions_{month}.csv"
return StreamingResponse(
iter([output.getvalue()]),
media_type="text/csv; charset=utf-8",
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
)
@router.get("/pdf")
async def export_pdf(
month: str = Query(default=None, description="Month YYYY-MM"),
session: AsyncSession = Depends(get_session),
current_user: User = Depends(get_current_user),
) -> StreamingResponse:
if month is None:
now = utcnow()
month = f"{now.year}-{now.month:02d}"
transactions = await _fetch_transactions(session, current_user.id, month)
total_income = sum(t.amount_cents for t in transactions if t.type.value == "income")
total_expense = sum(t.amount_cents for t in transactions if t.type.value == "expense")
rows_html = ""
for tx in transactions:
amount_eur = tx.amount_cents / 100
color = "#16a34a" if tx.type.value == "income" else "#dc2626"
sign = "+" if tx.type.value == "income" else "-"
rows_html += f"""
| {tx.transaction_date.isoformat()} |
{tx.category.name} |
{tx.description or "—"} |
{sign}{amount_eur:.2f} €
|
"""
html = f"""
Rapport de transactions
Période : {month} | {current_user.full_name or current_user.email}
Revenus
+{total_income/100:.2f} €
Dépenses
-{total_expense/100:.2f} €
Solde net
{(total_income - total_expense)/100:+.2f} €
| Date | Catégorie | Description | Montant |
{rows_html}
"""
import weasyprint # noqa: PLC0415
pdf_bytes = weasyprint.HTML(string=html).write_pdf()
filename = f"transactions_{month}.pdf"
return StreamingResponse(
iter([pdf_bytes]),
media_type="application/pdf",
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
)