How to Structure FastAPI Routers for PostGIS Tables

← Back to Spatial Resource Modeling Patterns

Isolate spatial endpoints into modular, domain-scoped router files, map geometry columns with GeoAlchemy2, and enforce strict Pydantic schemas that validate GeoJSON payloads before any data reaches the database.

# Context & When to Use

This approach is the right choice when a FastAPI application exposes more than one spatial entity—for example, parcels, sensors, and routes in the same codebase. Monolithic route files that mix spatial CRUD, analysis endpoints, and admin utilities become impossible to test in isolation and make it hard to attach entity-specific middleware (CRS validation, bounding-box sanitisation, rate limiting).

The pattern fits best when you are using SQLAlchemy 2.0 async with PostGIS and need geometry filtering to stay inside the database. If you are building a single-entity prototype or a read-only tile proxy, the added structure is premature—but for anything that will carry production traffic, separating router, schema, and query layers pays off quickly.

One precondition: every geometry column must carry an explicit SRID (typically EPSG:4326) and a GiST index. Without the index, ST_DWithin and ST_Intersects degrade to sequential scans that make router-level optimisations irrelevant.

# Recommended Directory Layout

The layout below separates concerns while keeping spatial logic explicit and testable. One router file per domain entity; Pydantic schemas and query services in parallel directories.

app/
├── routers/
│   ├── __init__.py
│   └── parcels.py          # Domain-scoped spatial router
├── models/
│   └── spatial.py          # SQLAlchemy + GeoAlchemy2 ORM
├── schemas/
│   └── spatial.py          # Pydantic v2 validation & serialisation
├── services/
│   └── spatial_queries.py  # Reusable PostGIS query functions
├── database.py             # Async engine & session factory
└── main.py                 # Router aggregation & app factory

This structure scales as your platform grows to include raster layers, topology checks, or multi-tenant spatial isolation. Each router imports only the models and schemas it requires, preventing circular dependencies and enabling independent deployment if you later migrate to microservices.

# Runnable Implementation

The four files below form a self-contained, production-ready setup for a parcels PostGIS table. Copy them in order; each step references the previous one.

Step 1 — ORM model (app/models/spatial.py)

from sqlalchemy import Column, Integer, String, CheckConstraint
from sqlalchemy.orm import DeclarativeBase
from geoalchemy2 import Geometry

class Base(DeclarativeBase):
    pass

class Parcel(Base):
    __tablename__ = "parcels"
    __table_args__ = (
        CheckConstraint("geom IS NOT NULL", name="parcels_geom_not_null"),
    )
    id   = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(100), nullable=False)
    # Explicit geometry type + SRID triggers automatic GiST index creation
    geom = Column(Geometry("POLYGON", srid=4326, spatial_index=True), nullable=False)

Step 2 — Pydantic v2 schemas (app/schemas/spatial.py)

from pydantic import BaseModel, Field, ConfigDict
from typing import Optional, Dict, Any

class ParcelCreate(BaseModel):
    name: str = Field(..., max_length=100)
    # Accept any valid GeoJSON geometry object; validated further in the router
    geom: Dict[str, Any] = Field(..., description="RFC 7946 GeoJSON geometry object")

class ParcelOut(BaseModel):
    id:   int
    name: str
    geom: Optional[Dict[str, Any]] = None
    model_config = ConfigDict(from_attributes=True)

For stricter RFC 7946 validation—enforcing correct coordinate ranges, geometry type constraints, and ring orientation—replace the Dict[str, Any] field with geojson-pydantic’s typed geometry models. See Strict Pydantic Validation for Geometry for a complete migration path.

Step 3 — Async database session (app/database.py)

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from typing import AsyncGenerator

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/gis_db"

engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,        # tune to CPU count × expected concurrent spatial queries
    max_overflow=10,
    echo=False,
)

AsyncSessionFactory = async_sessionmaker(engine, expire_on_commit=False)

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionFactory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

Step 4 — Spatial query service (app/services/spatial_queries.py)

from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.spatial import Parcel

async def get_parcels_within_radius(
    db:             AsyncSession,
    longitude:      float,
    latitude:       float,
    radius_meters:  float = 500.0,
    limit:          int   = 50,
) -> list[dict]:
    """
    ST_DWithin with geography casts uses metric distances (metres).
    Without the cast, the threshold would be interpreted as degrees—
    500 degrees matches the entire globe.
    """
    point_geog = func.ST_SetSRID(
        func.ST_MakePoint(longitude, latitude), 4326
    ).cast(func.geography())

    query = (
        select(Parcel.id, Parcel.name)
        .where(
            func.ST_DWithin(
                func.cast(Parcel.geom, func.geography()),
                point_geog,
                radius_meters,
            )
        )
        .limit(limit)
    )
    result = await db.execute(query)
    return [dict(row._mapping) for row in result.all()]

Step 5 — Domain-scoped router (app/routers/parcels.py)

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession
from shapely.geometry import shape
from geoalchemy2.elements import WKTElement

from app.models.spatial  import Parcel
from app.schemas.spatial import ParcelCreate, ParcelOut
from app.database        import get_db
from app.services.spatial_queries import get_parcels_within_radius

router = APIRouter(prefix="/parcels", tags=["parcels"])

@router.post("/", response_model=ParcelOut, status_code=201)
async def create_parcel(
    payload: ParcelCreate,
    db:      AsyncSession = Depends(get_db),
):
    try:
        shp = shape(payload.geom)          # validates geometry topology via Shapely
        db_parcel = Parcel(
            name=payload.name,
            geom=WKTElement(shp.wkt, srid=4326),
        )
        db.add(db_parcel)
        await db.commit()
        await db.refresh(db_parcel)
        return db_parcel
    except Exception as exc:
        await db.rollback()
        raise HTTPException(status_code=400, detail=f"Invalid geometry: {exc}")

@router.get("/nearby", response_model=list[dict])
async def list_nearby_parcels(
    lng:    float = Query(..., ge=-180, le=180),
    lat:    float = Query(..., ge=-90,  le=90),
    radius: float = Query(500.0, gt=0, description="Search radius in metres"),
    db:     AsyncSession = Depends(get_db),
):
    return await get_parcels_within_radius(db, lng, lat, radius)

Step 6 — App factory (app/main.py)

from fastapi import FastAPI
from app.routers import parcels

app = FastAPI(title="Geospatial Platform API", version="1.0.0")
# Versioned prefix isolates breaking changes from existing clients;
# see API Versioning for GIS Endpoints for migration strategies.
app.include_router(parcels.router, prefix="/api/v1")

Mount additional entity routers (sensors, zones, routes) with the same pattern: one include_router call per domain, each carrying its own prefix and tags. For a full strategy on evolving these prefixes without breaking existing clients, see API Versioning for GIS Endpoints.

# Key Parameters & Options

# Gotchas & Failure Modes

• Missing GiST index causes full table scans. GeoAlchemy2’s spatial_index=True creates the index for new tables, but existing tables need a manual migration: CREATE INDEX CONCURRENTLY idx_parcels_geom ON parcels USING GIST(geom);. Without it, ST_DWithin on a million-row table takes seconds instead of milliseconds.

• geography cast omitted in distance queries. ST_DWithin(geom, point, 500) with a geometry column uses degree-based distance. 500 degrees covers the entire globe. Always cast both arguments to geography when the threshold is in metres.

• WKTElement without SRID silently defaults to SRID 0. PostGIS stores the geometry but spatial index lookups that compare against SRID 4326 data return empty result sets. Always pass srid=4326 (or your target CRS) to WKTElement.

• Blocking ORM calls inside async handlers. Using synchronous psycopg2 sessions or calling session.execute() without await freezes the event loop under concurrent load. All SQLAlchemy calls inside async route handlers must use await.

• Lazy-loading N+1 on spatial relationships. If a Parcel has a relationship to Zone objects and you access parcel.zones inside a loop, SQLAlchemy fires one query per parcel. Use selectinload or an explicit JOIN with a single ST_Intersects predicate. The GeoJSON vs GeoParquet Serialization page covers serialisation strategies that compound with this problem for large response payloads.

# Verification

Confirm the setup is working with three quick checks:

1. SQL compilation check (unit test, no database needed)

from sqlalchemy.dialects import postgresql
from app.services.spatial_queries import get_parcels_within_radius
from sqlalchemy import select, func
from app.models.spatial import Parcel

query = (
    select(Parcel.id, Parcel.name)
    .where(func.ST_DWithin(
        func.cast(Parcel.geom, func.geography()),
        func.ST_SetSRID(func.ST_MakePoint(-0.1, 51.5), 4326).cast(func.geography()),
        500,
    ))
    .limit(50)
)
print(query.compile(dialect=postgresql.dialect(), compile_kwargs={"literal_binds": True}))
# Expect: ... WHERE ST_DWithin(CAST(parcels.geom AS geography), ...::geography, 500) ...

2. Live endpoint test

curl -s "http://localhost:8000/api/v1/parcels/nearby?lng=-0.1&lat=51.5&radius=1000" \
  | python3 -m json.tool
# Expect: a JSON array of {id, name} objects; empty array is fine if no data exists yet

3. Index usage via EXPLAIN

EXPLAIN (ANALYZE, BUFFERS)
SELECT id, name
FROM parcels
WHERE ST_DWithin(geom::geography, ST_SetSRID(ST_MakePoint(-0.1, 51.5), 4326)::geography, 1000);

The plan should show Index Scan using idx_parcels_geom (or the auto-generated name). A Seq Scan confirms the index is missing or the SRID mismatch is preventing its use.

# Related

• Spatial Resource Modeling Patterns — the parent page covering geometry type selection, connection pooling, and pagination design across all spatial entities
• Spatial Pagination & Cursor Strategies — replace offset pagination with bounding-box cursors that respect GiST indexes
• GeoJSON vs GeoParquet Serialization — format decision matrix for router response serialisation at scale
• API Versioning for GIS Endpoints — evolve router prefixes and schema versions without breaking existing clients
• Strict Pydantic Validation for Geometry — enforce RFC 7946 geometry constraints in the Pydantic layer before touching the database

← Back to Spatial Resource Modeling Patterns