import csv
class _LineBuffer:
def __init__(self):
self._data = ""
def write(self, value):
self._data += value
def take(self):
out, self._data = self._data, ""
return out
def iter_csv(rows, fieldnames):
buffer = _LineBuffer()
writer = csv.DictWriter(buffer, fieldnames=fieldnames, extrasaction="ignore")
writer.writeheader()
yield buffer.take().encode("utf-8")
for row in rows:
writer.writerow(row)
chunk = buffer.take()
if chunk:
yield chunk.encode("utf-8")
from contextlib import contextmanager
import psycopg2
import psycopg2.extras
@contextmanager
def fetch_rows(dsn, query, params=None, itersize=2000):
connection = psycopg2.connect(dsn)
try:
# A *named* cursor is a server-side cursor: rows are not all buffered
# client-side, which is what makes streaming actually stream.
cursor = connection.cursor(
name="csv_export",
cursor_factory=psycopg2.extras.RealDictCursor,
)
cursor.itersize = itersize
cursor.execute(query, params or {})
yield cursor
finally:
connection.close()
from datetime import date
from flask import Flask, Response, current_app, stream_with_context
from csv_stream import iter_csv
from db import fetch_rows
app = Flask(__name__)
FIELDS = ["id", "email", "plan", "created_at"]
EXPORT_QUERY = """
SELECT id, email, plan, created_at
FROM accounts
WHERE created_at >= %(since)s
ORDER BY id
"""
@app.route("/exports/accounts.csv")
def export_accounts():
dsn = current_app.config["DATABASE_DSN"]
params = {"since": date(date.today().year, 1, 1)}
def generate():
with fetch_rows(dsn, EXPORT_QUERY, params) as cursor:
for chunk in iter_csv(cursor, FIELDS):
yield chunk
filename = "accounts-%s.csv" % date.today().isoformat()
return Response(
stream_with_context(generate()),
mimetype="text/csv",
headers={"Content-Disposition": 'attachment; filename="%s"' % filename},
)
Exporting a large table as CSV over HTTP is a classic memory trap: building the whole file in a string or a list before returning it forces the entire dataset into RAM and blocks the worker until it finishes. This snippet shows the streaming alternative, where rows are produced lazily and flushed to the client in small chunks so memory stays flat regardless of table size.
In csv_stream.py, the core is iter_csv, a generator that yields already-encoded CSV byte chunks. Because Python's csv module writes to a file-like object, _LineBuffer acts as a tiny sink whose write just accumulates the current line; after each writerow the buffer is drained with take, so at most one row lives in memory at a time. The header row is emitted first, then rows arrive from an iterable of dicts. Yielding bytes (via .encode) matches what WSGI servers expect from a streaming response body.
The data source lives in db.py. fetch_rows opens a named cursor with connection.cursor(name=...), which in psycopg2 creates a server-side cursor. This is the key to true streaming from Postgres: without a name the driver buffers the full result set client-side, defeating the purpose. Setting itersize controls how many rows are fetched per network round trip, balancing round-trip overhead against memory. Iterating the cursor yields rows one batch at a time while the transaction stays open.
app.py wires it together. The route builds the row generator and hands iter_csv to Flask's Response with mimetype='text/csv'. Crucially the generator is wrapped in stream_with_context so the request and DB connection context remain alive while chunks are pulled lazily by the WSGI server. The Content-Disposition header prompts a file download, and a dated filename keeps exports distinguishable.
The main trade-off is that a streamed response can't easily change its status code or headers once bytes start flowing, so errors mid-stream are awkward — validation should happen before the first yield. It also holds a DB transaction open for the duration, so very slow clients can pin a connection. For bounded, download-style exports of large datasets this pattern keeps worker memory constant and lets users start receiving data immediately instead of waiting for the full file.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
class AddSettingsToAccounts < ActiveRecord::Migration[7.1]
disable_ddl_transaction!
def change
add_column :accounts, :settings, :jsonb, null: false, default: {}Postgres JSONB Partial Index for Feature Flags
package deps
import (
"net"
"net/http"
"time"HTTP client tuned for production: timeouts, transport, and connection reuse
Share this code
Here's the card — post it anywhere.
