Configuring GeoAlchemy2 Geometry Columns in FastAPI

This page covers the exact narrow problem of wiring a PostGIS geometry column through Model Mapping with GeoAlchemy2 into a FastAPI response — aligning three distinct layers that each fail silently when misconfigured: PostGIS spatial storage, SQLAlchemy 2.0 ORM type mapping, and Pydantic v2 JSON serialization.

Why the Naive Approach Fails

The most common mistake is declaring the geometry column without specifying geometry_type or srid, relying on GeoAlchemy2’s auto-created index, and then returning the ORM object directly from the endpoint:

python
# BROKEN: missing type/SRID, no serializer, naive column declaration
class Facility(Base):
    __tablename__ = "facilities"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    location: Mapped[Geometry] = mapped_column(Geometry())  # SRID defaults to 0

class FacilitySchema(BaseModel):
    id: int
    location: object  # FastAPI cannot serialize WKBElement

@app.get("/facilities/{id}")
def get_facility(id: int, db: Session = Depends(get_db)):
    return db.get(Facility, id)  # FastAPI raises TypeError on response build

Three things break here:

  1. SRID 0 — spatial functions like ST_DWithin and ST_Transform reject geometry without an explicit SRID, producing SRID mismatch (0 vs 4326) errors at query time rather than at insert time.
  2. WKBElement serialization — FastAPI’s JSON encoder raises TypeError: Object of type WKBElement is not JSON serializable because PostGIS returns geometry as raw binary (Well-Known Binary), not a Python dict.
  3. Uncontrolled index — Geometry() with defaults may create an unnamed GiST index you cannot reference in partial index definitions or composite spatial index strategies.

The architecture below shows how the three layers must connect:

GeoAlchemy2 FastAPI layer diagram Three connected layers: PostGIS storage emits WKBElement, GeoAlchemy2 ORM maps it to a Python object, Pydantic field_serializer converts it to a GeoJSON dict returned by FastAPI. PostGIS geometry column SRID 4326 · GiST → WKBElement GeoAlchemy2 ORM Geometry(type, srid) mapped_column → Python object FastAPI / Pydantic @field_serializer to_shape() → mapping() → GeoJSON dict WKB bytes WKBElement

Production-Ready Implementation

The following is a complete, copy-paste-ready setup. It maps a PostGIS POINT column, declares an explicit GiST index for spatial query performance, and serializes geometry safely for FastAPI endpoints.

python
from typing import Optional
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel, field_serializer
from sqlalchemy import create_engine, Integer, String, Index
from sqlalchemy.orm import DeclarativeBase, Session, mapped_column, Mapped
from geoalchemy2 import Geometry
from geoalchemy2.shape import to_shape
from geoalchemy2.elements import WKTElement
import shapely.geometry

DATABASE_URL = "postgresql+psycopg2://user:pass@localhost:5432/gis_db"
engine = create_engine(DATABASE_URL, echo=False)


class Base(DeclarativeBase):
    pass


# ── 1. SQLAlchemy model ───────────────────────────────────────────────────────
class Facility(Base):
    __tablename__ = "facilities"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(100), nullable=False)

    # Explicit geometry type + SRID prevents silent SRID=0 mismatch errors.
    # spatial_index=False lets us declare the index explicitly below.
    location: Mapped[Geometry] = mapped_column(
        Geometry(geometry_type="POINT", srid=4326, spatial_index=False)
    )

    __table_args__ = (
        # Named index: easier to reference in EXPLAIN output and partial-index
        # definitions. Use postgresql_using="gist" — B-tree cannot handle bbox ops.
        Index("idx_facilities_location_gist", "location", postgresql_using="gist"),
    )


# ── 2. Pydantic schema with GeoJSON serialization ─────────────────────────────
class FacilitySchema(BaseModel):
    id: int
    name: str
    location: Optional[dict] = None

    # Required for SQLAlchemy ORM object compatibility in Pydantic v2.
    model_config = {"from_attributes": True}

    @field_serializer("location")
    def serialize_geometry(self, value: Optional[object]) -> Optional[dict]:
        if value is None:
            return None
        # WKBElement → Shapely geometry → GeoJSON dict.
        # Never pass WKBElement directly to json.dumps() — it raises TypeError.
        geom = to_shape(value)
        return shapely.geometry.mapping(geom)


# ── 3. FastAPI app and endpoints ──────────────────────────────────────────────
app = FastAPI(title="GeoAlchemy2 + FastAPI")


def get_db():
    with Session(engine) as session:
        yield session


@app.post("/facilities/", response_model=FacilitySchema, status_code=201)
def create_facility(name: str, lat: float, lon: float, db: Session = Depends(get_db)):
    # WKTElement attaches the SRID before PostGIS receives the geometry,
    # preventing the "Geometry SRID (0) does not match column SRID (4326)" error.
    wkt_point = WKTElement(f"POINT({lon} {lat})", srid=4326)
    facility = Facility(name=name, location=wkt_point)
    db.add(facility)
    db.commit()
    # db.refresh() pulls the server-generated id and persisted geometry back
    # into the instance. Without it, location may be stale or None in the response.
    db.refresh(facility)
    return facility


@app.get("/facilities/{facility_id}", response_model=FacilitySchema)
def get_facility(facility_id: int, db: Session = Depends(get_db)):
    facility = db.get(Facility, facility_id)
    if not facility:
        raise HTTPException(status_code=404, detail="Facility not found")
    return facility

Configuration and Tuning

GiST index strategy. Pair this setup with the GiST index optimization patterns when your facilities table grows beyond a few thousand rows. For workloads that query only a subset of rows — for example, only active facilities — convert the index to a partial index:

python
Index(
    "idx_active_facilities_gist",
    "location",
    postgresql_using="gist",
    postgresql_where="status = 'active'",
)

This reduces index size and write amplification for INSERT-heavy pipelines. See creating partial indexes for active map regions for the full pattern including ANALYZE requirements.

SQLAlchemy connection pool. The default pool size of 5 connections is sufficient for development but undersized for concurrent spatial workloads. Tune with:

python
engine = create_engine(
    DATABASE_URL,
    pool_size=10,          # persistent connections
    max_overflow=5,        # burst headroom
    pool_pre_ping=True,    # detect stale connections before use
)

work_mem for spatial sorts. ST_DWithin radius searches that order results by distance may spill to disk if work_mem is too low. Set it per-session for spatial endpoints:

python
db.execute(text("SET LOCAL work_mem = '32MB'"))

Output format. The implementation returns GeoJSON, which is the standard consumed by Leaflet, MapLibre, and OpenLayers. If a downstream service expects Well-Known Text instead, replace shapely.geometry.mapping(geom) with geom.wkt and change the Pydantic field type from dict to str.

Verification Steps

After running Base.metadata.create_all(engine), confirm the index was created and the planner will use it:

sql
-- Confirm GiST index exists
SELECT indexname, indexdef
FROM pg_indexes
WHERE tablename = 'facilities'
  AND indexname = 'idx_facilities_location_gist';

-- Insert a test row (adjust connection details)
INSERT INTO facilities (name, location)
VALUES ('Test', ST_SetSRID(ST_MakePoint(-73.985, 40.748), 4326));

-- Verify the planner uses the index for a spatial predicate
EXPLAIN (ANALYZE, BUFFERS)
SELECT id, name
FROM facilities
WHERE ST_DWithin(
    location,
    ST_SetSRID(ST_MakePoint(-73.985, 40.748), 4326)::geography,
    1000
);
-- Expected: "Index Scan using idx_facilities_location_gist on facilities"
-- If you see "Seq Scan", run ANALYZE facilities; then re-check.

Verify the FastAPI serializer from Python:

python
import requests

# POST a facility
r = requests.post(
    "http://localhost:8000/facilities/",
    params={"name": "Central Park", "lat": 40.785, "lon": -73.968}
)
assert r.status_code == 201
body = r.json()
assert body["location"]["type"] == "Point"
assert len(body["location"]["coordinates"]) == 2
print(body["location"])
# {"type": "Point", "coordinates": [-73.968, 40.785]}

Reading the EXPLAIN ANALYZE output for spatial joins covers how to interpret cost estimates and bitmap heap scans when the index is used.

Gotchas Checklist

  • Omitting srid in Geometry() defaults the column to SRID 0. Spatial functions reject cross-SRID operations silently or with cryptic errors at query time. Always pass the target SRID explicitly.
  • Skipping db.refresh() after commit leaves the ORM instance holding only the client-side values supplied during INSERT. The location field will be the raw WKTElement string you passed in, not the fully resolved WKBElement the serializer expects — producing a to_shape() AttributeError.
  • Using spatial_index=True (the default) alongside an explicit Index(...) creates two indexes on the same column: one unnamed (from GeoAlchemy2) and one named (from your declaration). The duplicate wastes storage and write overhead. Set spatial_index=False whenever you declare your own index.
  • Autocommit mode with psycopg2 bypasses SQLAlchemy’s transaction scope. If you switch to AUTOCOMMIT isolation for bulk inserts, remove db.commit() calls and use explicit BEGIN/COMMIT in raw SQL or switch to engine.begin() context blocks.
  • SRID cast overhead in hot paths. If PostGIS must cast the input geometry’s SRID to match the column on every row during a bulk insert, query latency spikes. Batch inserts should attach the correct SRID at construction time (WKTElement(..., srid=4326)) rather than relying on server-side ST_SetSRID calls in a trigger.