Migrating Spatial Code From psycopg2 to psycopg3

This walkthrough ports a running GeoAlchemy2 vessel-tracking backend from the legacy driver to the rewritten one, extending the driver comparison in Choosing a PostGIS Driver into a concrete, ordered migration. The service ingests ais_positions (a geometry(Point,4326) column keyed by MMSI) and serves port_zones polygons over a FastAPI API. The goal is to swap postgresql+psycopg2 for postgresql+psycopg without touching a single ORM model, so the risk is contained to a handful of raw-driver call sites.

Why the naive approach fails

The tempting shortcut is to only edit the connection string and redeploy. That works for pure-ORM code, but any module that reached around SQLAlchemy into psycopg2 directly will break on import, because the fast-path helpers moved or vanished. A typical AIS ingester carries exactly that kind of code:

python
# BREAKS under psycopg 3: execute_values no longer exists
from psycopg2.extras import execute_values   # ImportError on psycopg 3
import psycopg2

def ingest(dsn, batch):
    conn = psycopg2.connect(dsn)              # psycopg2.connect is gone too
    cur = conn.cursor()
    execute_values(
        cur,
        "INSERT INTO ais_positions (mmsi, recorded_at, geom) VALUES %s",
        batch,
        template="(%s, to_timestamp(%s), ST_SetSRID(ST_MakePoint(%s,%s), 4326))",
    )
    conn.commit()

Change only the engine URL and this module still imports psycopg2.extras, so the process fails at startup. The migration therefore has to find every direct psycopg2 reference, not just the SQLAlchemy URL.

Production-ready implementation

Work in five ordered steps. Steps 1 through 4 are the code edits; step 5 is verification.

Step 1 — Install psycopg 3 and switch the dialect URL

Add the driver with its binary and pool extras, then change the one URL. Keep psycopg2-binary installed until cutover so you can roll back instantly.

text
# requirements.txt
sqlalchemy>=2.0.15
geoalchemy2>=0.14
psycopg[binary,pool]>=3.1
psycopg2-binary>=2.9   # keep during migration; remove after cutover
python
from sqlalchemy import create_engine

# BEFORE: engine = create_engine("postgresql+psycopg2://ais:secret@db/ais_tracking")
engine = create_engine(
    "postgresql+psycopg://ais:secret@db.internal:5432/ais_tracking",
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True,
    pool_recycle=1800,
)

The dialect name is psycopg — not psycopg3. A postgresql+psycopg3:// URL raises NoSuchModuleError.

Step 2 — Replace execute_values with executemany plus pipeline

psycopg removed execute_values. The direct replacement is cursor.executemany() with a single-row template, wrapped in a conn.pipeline() block so the round-trips are batched. This is the whole ingest path rewritten and ready to paste in:

python
import psycopg

DSN = "postgresql://ais:secret@db.internal:5432/ais_tracking"

INSERT_SQL = (
    "INSERT INTO ais_positions (mmsi, recorded_at, geom) "
    "VALUES (%s, to_timestamp(%s), ST_SetSRID(ST_MakePoint(%s, %s), 4326))"
)

def ingest(rows: list[tuple]) -> int:
    """rows: (mmsi, epoch_seconds, lon, lat). SRID 4326 attached in SQL."""
    with psycopg.connect(DSN) as conn:      # psycopg.connect, context-managed
        with conn.pipeline():               # collapse N round-trips into one flush
            with conn.cursor() as cur:
                cur.executemany(INSERT_SQL, rows)
        conn.commit()
    return len(rows)

For the very largest loads, psycopg’s native COPY beats even pipelined inserts. Stream WKB straight in, one row at a time, without building a giant SQL string:

python
import psycopg
from shapely.geometry import Point
from shapely import wkb as shp_wkb

def copy_positions(rows: list[tuple]) -> None:
    """rows: (mmsi, epoch_seconds, lon, lat) — bulk load via COPY."""
    with psycopg.connect(DSN) as conn:
        with conn.cursor() as cur:
            with cur.copy(
                "COPY ais_positions (mmsi, recorded_at, geom) "
                "FROM STDIN (FORMAT BINARY)"
            ) as copy:
                copy.set_types(["int8", "timestamptz", "geometry"])
                for mmsi, epoch, lon, lat in rows:
                    from datetime import datetime, timezone
                    ts = datetime.fromtimestamp(epoch, tz=timezone.utc)
                    # WKB with SRID 4326 embedded so PostGIS stores it directly
                    geom = shp_wkb.dumps(Point(lon, lat), srid=4326, hex=False)
                    copy.write_row((mmsi, ts, geom))
        conn.commit()

Step 3 — Update server-side cursor usage

If your export path used psycopg2 named cursors directly (conn.cursor(name="stream")), move it up to the SQLAlchemy layer, where stream_results and yield_per request a server-side cursor portably. This code runs unchanged on either driver, which is exactly why it is the right layer to own the behaviour:

python
from sqlalchemy import select
from geoalchemy2.shape import to_shape
from models import AisPosition

def export_track(engine, mmsi: int, batch: int = 2000):
    stmt = (
        select(AisPosition.recorded_at, AisPosition.geom)
        .where(AisPosition.mmsi == mmsi)
        .order_by(AisPosition.recorded_at)
        .execution_options(stream_results=True, yield_per=batch)
    )
    with engine.connect() as conn:
        for row in conn.execute(stmt):
            pt = to_shape(row.geom)         # WKBElement -> shapely Point (4326)
            yield row.recorded_at.isoformat(), pt.x, pt.y

Step 4 — Port adapters and placeholders

Two smaller concerns close out the code changes. First, both drivers accept pyformat placeholders (%s), so parameterised SQL binds unchanged — do not convert them to anything else. Second, any custom psycopg2.extensions.register_adapter call must be rewritten against psycopg 3’s psycopg.adapt system. Most AIS backends only adapt scalars and can drop the custom adapters entirely, letting geometry travel as WKB through GeoAlchemy2. If you passed Shapely objects straight into raw SQL under psycopg2, convert them explicitly instead:

python
from shapely import wkb as shp_wkb
from shapely.geometry import Point

# Convert Shapely -> WKB bytes and let ST_GeomFromWKB attach the SRID
point_wkb = shp_wkb.dumps(Point(4.4792, 51.9080), srid=4326, hex=False)
# ... "INSERT INTO port_zones (geom) VALUES (ST_GeomFromWKB(%s, 4326))", (point_wkb,)

Step 5 — Verify

Round-trip a known coordinate and confirm the plan still uses the GiST index. If both pass, the cutover is safe.

python
from sqlalchemy import text
from shapely import wkb as shp_wkb

def verify(engine) -> dict:
    with engine.connect() as conn:
        wkb = conn.execute(text(
            "SELECT ST_AsBinary(ST_SetSRID(ST_MakePoint(4.4792, 51.9080), 4326))"
        )).scalar()
    pt = shp_wkb.loads(bytes(wkb))
    return {
        "driver": engine.dialect.driver,          # expect 'psycopg'
        "lon": round(pt.x, 4),                     # 4.4792
        "lat": round(pt.y, 4),                     # 51.9080
    }
sql
-- Confirm the migrated engine still hits the spatial index
EXPLAIN (ANALYZE, BUFFERS)
SELECT mmsi
FROM   ais_positions
WHERE  ST_DWithin(
           geom::geography,
           ST_SetSRID(ST_MakePoint(4.4792, 51.9080), 4326)::geography,
           2000        -- metres from the Rotterdam approach
       );
-- Expect: Index Scan using ais_positions_geom_gist

Configuration and tuning knobs

  • pool_recycle — keep 1800 s; psycopg 3 pooling semantics match psycopg2 here, and long-lived AIS ingest connections still benefit from periodic recycling.
  • prepare_threshold — psycopg 3 auto-prepares hot statements. Behind PgBouncer transaction pooling this collides; set postgresql+psycopg://...?prepare_threshold=0 or use a session-pooling port.
  • Pipeline batch size — executemany inside conn.pipeline() benefits from feeding a few thousand rows per flush. Tune to your AIS message rate; oversized batches raise peak memory without cutting latency further.
  • yield_per — 1000 to 5000 for the export path. This sets the server-side cursor fetch size; too small adds round-trips, too large defeats the streaming memory saving.
  • work_mem — unchanged by the driver swap; keep it sized for any ORDER BY ST_Distance sorts in your export queries.

Verification steps

  1. Run the verify() helper and assert driver == "psycopg" with coordinates returning 4.4792, 51.9080.
  2. Diff row counts before and after a test ingest to confirm executemany/COPY inserted every AIS point.
  3. Run the EXPLAIN (ANALYZE, BUFFERS) above and confirm Index Scan, not Seq Scan.
  4. Load-test the pipeline path against the old execute_values path and record the round-trip reduction.
  5. Once green, remove psycopg2-binary from requirements and redeploy to prove no residual imports remain.

Gotchas checklist

  • The dialect is psycopg, never psycopg3. The URL prefix postgresql+psycopg3 raises NoSuchModuleError.
  • execute_values is gone. Replace it with executemany in a pipeline or with copy(); do not vendor the old helper.
  • Direct psycopg2 imports break at startup, not at runtime. Grep the codebase for import psycopg2 and psycopg2.extras before deploying — a missed one crashes the process on boot.
  • Prepared-statement collisions behind PgBouncer. psycopg 3 auto-prepares; add prepare_threshold=0 when transaction pooling is in play.
  • Do not change SRID handling during the migration. Keep ST_SetSRID(..., 4326) and ST_GeomFromWKB(%s, 4326) exactly as they were; conflating the driver swap with an SRID change makes failures impossible to bisect.