Async Streaming of Large Geometry Result Sets

Exporting a day of AIS traffic is where Session Management for Spatial Data meets its hardest test: an ais_positions table holding tens of millions of geometry(Point,4326) rows must flow to an HTTP client as GeoJSON without ever loading the whole set into memory. Doing this on a modern async FastAPI stack means combining SQLAlchemy’s AsyncSession, an async server-side cursor, stream_results with yield_per, and backpressure from the response, so memory stays flat while millions of geometries stream out one feature at a time.

Why the naive approach fails

The default await session.execute(stmt) buffers the entire result before returning. On a large export that materialises every row and every WKBElement at once, spiking memory until the worker is killed:

python
# OOM: buffers the whole day of positions before returning a single byte
async def export_day_broken(session, day):
    result = await session.execute(
        select(AisPosition).where(AisPosition.recorded_at >= day)
    )
    rows = result.scalars().all()            # tens of millions of objects
    return [to_shape(r.geom).__geo_interface__ for r in rows]

Two things go wrong. First, .all() defeats streaming entirely — it reads the whole cursor. Second, even if you iterated, a plain execute on an async engine still buffers unless you explicitly request a server-side cursor. The fix is to switch to session.stream(...) with stream_results set, and to push serialization into SQL so no per-row WKBElement is built.

Production-ready implementation

The complete pattern below wires an async engine on psycopg 3, opens an async server-side cursor with yield_per, serializes each row with ST_AsGeoJSON in the database, and streams newline-delimited GeoJSON features out of a FastAPI endpoint under natural backpressure.

python
from datetime import datetime, timezone
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from sqlalchemy import BigInteger, DateTime, Integer, func, select
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from geoalchemy2 import Geometry
import json


class Base(DeclarativeBase):
    pass


class AisPosition(Base):
    __tablename__ = "ais_positions"
    id:   Mapped[int] = mapped_column(Integer, primary_key=True)
    mmsi: Mapped[int] = mapped_column(BigInteger, index=True)
    recorded_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True)
    geom: Mapped[object] = mapped_column(
        Geometry(geometry_type="POINT", srid=4326, spatial_index=True)
    )


# psycopg 3 is required: it supports async server-side cursors AND decodes
# PostGIS geometry through GeoAlchemy2. asyncpg cannot do both.
engine = create_async_engine(
    "postgresql+psycopg://ais:secret@db.internal:5432/ais_tracking",
    pool_size=5,
    max_overflow=10,
    pool_pre_ping=True,
    connect_args={
        # server-side settings applied per connection
        "options": "-c statement_timeout=0 -c work_mem=64MB",
    },
)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)


async def stream_positions_geojson(day_start: datetime):
    """Yield GeoJSON Feature lines for one day of AIS points, memory-flat."""
    stmt = (
        select(
            AisPosition.mmsi,
            AisPosition.recorded_at,
            # Serialize in the database — precision 6 (~0.1 m) keeps lines small.
            func.ST_AsGeoJSON(AisPosition.geom, 6).cast(JSONB).label("geometry"),
        )
        .where(AisPosition.recorded_at >= day_start)
        .order_by(AisPosition.recorded_at)
        # stream_results opens an async server-side cursor;
        # yield_per bounds each fetch to 2000 rows.
        .execution_options(stream_results=True, yield_per=2000)
    )
    async with AsyncSessionLocal() as session:
        result = await session.stream(stmt)      # NOT execute — stream()
        async for row in result:
            feature = {
                "type": "Feature",
                "properties": {"mmsi": row.mmsi, "t": row.recorded_at.isoformat()},
                "geometry": row.geometry,
            }
            # newline-delimited GeoJSON: one feature per line, flushable
            yield json.dumps(feature, separators=(",", ":")) + "\n"


app = FastAPI()


@app.get("/exports/positions")
async def export_positions(day: str):
    day_start = datetime.fromisoformat(day).replace(tzinfo=timezone.utc)
    # StreamingResponse awaits each chunk, so the cursor only advances as fast
    # as the client drains the socket — this is the backpressure mechanism.
    return StreamingResponse(
        stream_positions_geojson(day_start),
        media_type="application/geo+json",
    )

The three moving parts each carry their weight. session.stream() (rather than execute) plus stream_results=True opens an async server-side cursor on psycopg 3, so PostgreSQL holds the result and ships it in slices. yield_per=2000 sets the fetch size, bounding how many rows cross the wire per round-trip. And because StreamingResponse awaits each yielded chunk, the async generator suspends whenever the client is slow to read — the database is never pushed to produce faster than the consumer drains, which is exactly backpressure.

Pushing ST_AsGeoJSON into the SELECT list is what keeps this both fast and lean: PostgreSQL builds each feature’s JSON once and streams text, so no WKBElement is ever constructed in Python. If a downstream job needs Shapely geometry instead, swap the column for ST_AsBinary(geom) and decode with shapely.wkb.loads inside the loop — the streaming skeleton is identical.

Configuration and tuning knobs

  • yield_per — the server-side cursor fetch size. 1000 to 5000 suits AIS point exports; smaller adds round-trips, larger raises the per-batch memory floor. This is the single most important streaming knob.
  • stream_results=True — mandatory. Without it, even session.stream on some paths can buffer; set it explicitly on the statement’s execution_options.
  • statement_timeout=0 — a full-day export legitimately runs for minutes. Disable the per-statement timeout for this connection only (via connect_args), never globally, and keep the OLTP engine’s timeout tight.
  • pool_size — keep the streaming engine’s pool small (5 or so). Long-lived export cursors each pin a connection, so a large pool invites exhaustion under concurrent exports.
  • work_mem — the ORDER BY recorded_at sort benefits from adequate work_mem; 64 MB avoids a disk spill on the sort while the stream runs. Better still, add a composite (recorded_at) or (mmsi, recorded_at) index so the order comes from the index and the sort disappears.
  • Precision on ST_AsGeoJSON — 6 decimal places is ~0.1 m for lon/lat, plenty for vessel positions and much smaller on the wire than the default 9.

Verification steps

Confirm the export holds memory flat regardless of row count. Sample resident memory as the stream runs and assert it does not grow with the number of features emitted:

python
import asyncio
from datetime import datetime, timezone
import resource

async def profile_export():
    day = datetime(2026, 7, 1, tzinfo=timezone.utc)
    count = 0
    peak_start = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
    async for _line in stream_positions_geojson(day):
        count += 1
        if count % 100_000 == 0:
            rss = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
            print(f"{count:>9} features, RSS {rss // 1024} MB")
    print("total features:", count, "start RSS KB:", peak_start)

asyncio.run(profile_export())
# RSS should plateau early and stay flat as feature count climbs into millions.

Then confirm the query plan streams from the index rather than sorting the whole table:

sql
-- Expect an Index Scan feeding the stream, no full-table Sort node
EXPLAIN (ANALYZE, BUFFERS)
SELECT mmsi, recorded_at, ST_AsGeoJSON(geom, 6)
FROM   ais_positions
WHERE  recorded_at >= timestamptz '2026-07-01 00:00:00+00'
ORDER  BY recorded_at;
-- If a Sort node appears, add an index on recorded_at so the order is free.

A healthy result shows RSS plateauing within the first few batches and holding steady into the millions of features, plus an Index Scan (or index-ordered scan) with no large Sort node. Rising memory means a buffering call slipped in — check that .all() is absent and that session.stream is used in place of session.execute.

Gotchas checklist

  • session.execute(...).all() defeats streaming. It reads the entire cursor into memory. Use session.stream(...) and iterate with async for.
  • stream_results is required even on an async engine. A bare async execute may still buffer; set stream_results=True (and yield_per) on the statement.
  • asyncpg cannot decode geometry through GeoAlchemy2. For an async ORM stream of geometry use postgresql+psycopg; if you must use asyncpg, select ST_AsGeoJSON/ST_AsBinary and decode manually.
  • A global statement_timeout kills long exports. Relax it on the streaming connection only, and keep transactional connections strictly capped.
  • No index on the ORDER BY column forces a full sort. Sorting the whole day before the first row streams defeats the memory goal; index recorded_at so the stream starts immediately.
  • Large connection pools plus long cursors exhaust connections. Each open export holds a connection for its whole duration; keep the streaming engine’s pool small and separate from the OLTP pool.