fastapi-csv-export-streaming

Minimal FastAPI service that streams leads as CSV with date-range and optional status filters.

Stack

• FastAPI – web framework
• SQLAlchemy 2 – ORM
• SQLite – database (default)
• Pydantic – validation and settings
• Uvicorn – ASGI server

Setup

1. Create a virtual environment and activate it:

   python -m venv .venv
   .venv\Scripts\activate   # Windows
   # source .venv/bin/activate   # Linux/macOS

2. Install dependencies:

   pip install -r requirements.txt

3. (Optional) Set DATABASE_URL in a .env file. Default is sqlite:///./leads.db.

Run

From the project root:

uvicorn app.main:app --reload

• API: http://127.0.0.1:8000
• Docs: http://127.0.0.1:8000/docs

On first startup, the app creates the SQLite DB, tables, and seeds 50 sample leads.

Export endpoint

GET /export/leads.csv

Parameter	Type	Required	Description
start_date	string	Yes	Start date (YYYY-MM-DD)
end_date	string	Yes	End date (YYYY-MM-DD)
status	string	No	Filter by lead status

• Date range: start_date is inclusive (00:00:00); end_date is exclusive (range ends at next day 00:00:00).
• Streaming: Response is streamed as CSV; no full load of rows into memory.

Example curl

curl -o leads.csv "http://127.0.0.1:8000/export/leads.csv?start_date=2025-02-01&end_date=2025-03-01"

With optional status filter:

curl -o leads.csv "http://127.0.0.1:8000/export/leads.csv?start_date=2025-02-01&end_date=2025-03-01&status=new"

Project layout

fastapi-csv-export-streaming/
├── app/
│   ├── __init__.py
│   ├── config.py       # Settings (e.g. DATABASE_URL)
│   ├── database.py     # Engine, session, get_db
│   ├── main.py         # FastAPI app, lifespan, router
│   ├── models.py       # Lead model
│   ├── routers/
│   │   ├── __init__.py
│   │   └── export.py   # GET /export/leads.csv
│   └── seed.py         # Seed 50 sample leads
├── requirements.txt
├── README.md
└── manual_test.md