python 41 lines · 7 steps

Streaming Postgres rows with a server-side cursor

A named server-side cursor lets you iterate a huge result set in fixed batches without loading it all into memory.

Explained by highlit

Intermediate generators context-managers database-streaming resource-cleanup memory-efficiency

1from contextlib import contextmanager
2from typing import Iterator
3 
4import psycopg2
5import psycopg2.extras
6 
7 
8@contextmanager
9def server_side_cursor(dsn: str, name: str, itersize: int = 2000):
10    conn = psycopg2.connect(dsn)
11    try:
12        cursor = conn.cursor(name=name, cursor_factory=psycopg2.extras.RealDictCursor)
13        cursor.itersize = itersize
14        try:
15            yield cursor
16        finally:
17            cursor.close()
18    finally:
19        conn.close()
20 
21 
22def stream_orders(dsn: str, since: str, batch_size: int = 2000) -> Iterator[dict]:
23    query = """
24        SELECT id, customer_id, total_cents, placed_at
25        FROM orders
26        WHERE placed_at >= %s
27        ORDER BY placed_at
28    """
29    with server_side_cursor(dsn, "orders_stream", batch_size) as cursor:
30        cursor.execute(query, (since,))
31        for row in cursor:
32            yield row
33 
34 
35def export_revenue_by_customer(dsn: str, since: str) -> dict[int, int]:
36    totals: dict[int, int] = {}
37    for order in stream_orders(dsn, since):
38        totals[order["customer_id"]] = (
39            totals.get(order["customer_id"], 0) + order["total_cents"]
40        )
41    return totals

01 / 01

STEP 01

Walkthrough

Space play ←→ step click any line

Three takeaways

1Named cursors keep the result set on the server so the client fetches rows in bounded batches.
2Nested try/finally blocks guarantee both cursor and connection close even when iteration fails midway.
3Chaining generators lets you aggregate an arbitrarily large dataset with constant memory.

Related explainers

python

import wave
import os
from dataclasses import dataclass

Reading WAV metadata into a dataclass

dataclass audio file-io

Beginner 5 steps

python

from pathlib import Path
from collections import defaultdict
 
from PIL import Image

Finding near-duplicate images by perceptual hash

perceptual-hashing clustering hamming-distance

Intermediate 9 steps

python

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse, Response
from starlette.status import HTTP_413_REQUEST_ENTITY_TOO_LARGE

Enforcing a max request body size in FastAPI

middleware streaming request-limits

Advanced 6 steps

python

import logging
import uuid
from contextvars import ContextVar

Request ID tracing in FastAPI middleware

middleware context-variables request-tracing

Intermediate 7 steps

python

import json
import time
import queue

Server-Sent Events streaming in Flask

server-sent-events streaming pub-sub

Advanced 9 steps

python

from flask import Blueprint, request, jsonify
from marshmallow import Schema, fields, validate, ValidationError, EXCLUDE
 
from .models import db, User