Add a CSV export for invoices
Python · FastAPI · advanced · modification
Adds the invoice CSV export requested for the billing dashboard. `app/routers/exports.py` introduces GET /exports/invoices.csv, which authenticates the caller, pulls the invoices for the requested status, and streams them back as a downloadable CSV via `StreamingResponse`. `app/deps.py` (the `current_org_member` dependency) and the existing `app/routers/invoices.py` list endpoint are shown unchanged for reviewer context. Tested against the billing dashboard: the download opens in a spreadsheet and the status filter works.
Multi-tenant SaaS: a user can belong to several orgs, and every request carries an `X-Org-Id` header. `current_org_member` (app/deps.py) is the audited boundary that maps an authenticated user + `X-Org-Id` to a Membership; `member.org_id` is what every org-scoped query filters on. `get_current_user` only proves the request is authenticated, not which org's data it may see. Invoice rows from all orgs live in one `invoices` table keyed by `org_id`.
Requirements
- Add GET /exports/invoices.csv: a signed-in user downloads their org's invoices (number, issue date, total, status) as a CSV file, optionally filtered by status.
- Every endpoint that returns org-owned data must scope its query to the caller's org through the documented tenant-isolation dependency (current_org_member) and filter on member.org_id -- exactly as the existing GET /invoices does. Authentication alone is not scoping.
- current_org_member is the tenant-isolation boundary (see its docstring): an org-scoped query that does not filter on member.org_id returns other tenants' rows and is a defect even if the caller is authenticated.
- The export streams a real CSV (header row + one row per invoice) and reuses the same invoice fields the list endpoint exposes.
Files touched
- app/deps.py
- app/routers/invoices.py
- app/routers/exports.py
--- app/deps.py
from fastapi import Depends, Header, HTTPException, status
from sqlalchemy.orm import Session
from app.database import get_db
from app.models import Membership, User
from app.security import get_current_user
def current_org_member(
user: User = Depends(get_current_user),
x_org_id: int = Header(..., alias="X-Org-Id"),
db: Session = Depends(get_db),
) -> Membership:
"""Resolve the caller's membership in the requested org.
TENANT-ISOLATION BOUNDARY. This is the one dependency that binds a request
to exactly one org: it verifies the authenticated user really belongs to
`X-Org-Id` and returns their Membership. The returned `member.org_id` is
the ONLY value an org-scoped query may filter on.
Every endpoint that reads or writes org-owned data MUST depend on this and
MUST filter by `member.org_id`. Authentication alone is not enough -- a
query that skips this scope returns other tenants' rows.
"""
membership = (
db.query(Membership)
.filter(
Membership.user_id == user.id,
Membership.org_id == x_org_id,