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

Four API changes that break spatial code silently Four rows comparing psycopg2 and psycopg3. The with-connection block now commits and closes rather than only committing. Server-side cursors are created differently. Type adapters move to a per-connection registry. And the geometry adapter registration point changes, which is where WKB handling usually breaks. The differences that bite spatial code specifically connection context manager with conn: → commits only with conn: → commits AND closes server-side cursor conn.cursor(name='c') conn.cursor(name='c') + row_factory type adaptation register_adapter → conn.adapters registry A geometry adapter registered globally in psycopg2 style is silently ignored — geometries come back as strings.

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

Migrating one service at a time, with both drivers installed Four stages. Install psycopg3 alongside psycopg2. Point read-only endpoints at the new dialect URL and compare results. Move writes once reads are proven. Remove psycopg2 from the dependency list only after a full release cycle with no rollbacks. Both drivers can be installed at once — use that 1 · install both pip install psycopg no code changes yet 2 · reads first +psycopg on replicas compare geometry output 3 · then writes watch transaction scope and adapter registration 4 · remove psycopg2 a cycle later Only the dialect prefix changes in the URL: postgresql+psycopg2 becomes postgresql+psycopg. Everything above SQLAlchemy stays the same, which is what makes a per-service rollout practical. Keep the old dialect URL behind an environment variable so a rollback is a config change, not a deploy.

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

Compare the bytes, not the rendered output The same row read through both drivers. The test hashes the WKB bytes returned by each and asserts they are identical, which catches encoding differences that a visual comparison of coordinates would miss. The acceptance test is a hash comparison psycopg2 path hex str → unhexlify → bytes sha256 = 4f2a…c918 psycopg3 path binary → bytes sha256 = 4f2a…c918 identical Run it across a sample that includes an empty geometry, a geometry with Z values and one large enough to be TOASTed — those are where encoding differences surface, not on a simple point.
  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.