Spatial Resource Modeling Patterns

← Back to Core Geospatial API Architecture

Designing production-grade geospatial APIs requires more than mapping database tables to JSON endpoints. This page establishes repeatable conventions for representing geographic entities, managing coordinate reference systems, optimizing spatial queries, and structuring API boundaries in FastAPI with PostGIS — the decisions that determine whether your service scales predictably or collapses under concurrent load.

# Prerequisites & Environment

Before implementing these patterns, confirm your stack meets the following baseline:

• fastapi>=0.100 with asyncio-compatible database drivers (asyncpg>=0.29)
• sqlalchemy>=2.0 with geoalchemy2>=0.14 for PostGIS type mapping
• PostgreSQL 14+ with PostGIS 3.3+ — verify with SELECT PostGIS_Full_Version();
• pydantic>=2.0 for field validators and model serialization
• Familiarity with OGC Simple Features geometry types and coordinate reference systems

# Decision Matrix: Geometry Type Selection

This is the first modeling decision you make and the one that most affects query performance and API contract stability. Choose incorrectly and you get silent precision loss or severe latency spikes.

When modeling resources that span regional or global extents, default to the spheroidal geography type. For localized, high-throughput applications, planar geometry with an explicit SRID constraint outperforms at scale.

# Step-by-Step Implementation

# Step 1: Define Spatial Column Types with Enforced Constraints

Declare the column type explicitly in your SQLAlchemy model. The SRID and spatial index must be co-defined with the column — post-hoc migration of SRID constraints is error-prone and may silently accept mismatched incoming coordinates.

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__ = (
        # Reject null geometries at the DB level — not just the application layer
        CheckConstraint("geom IS NOT NULL", name="parcels_geom_not_null"),
    )

    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    # geometry_type constrains the column; spatial_index=True creates the GIST index
    geom = Column(
        Geometry(geometry_type="POLYGON", srid=4326, spatial_index=True)
    )

Document the expected coordinate order (longitude, then latitude, per RFC 7946) in your OpenAPI schema. Failing to do so causes client-side inversion bugs that produce geometries mirrored across the prime meridian — a notoriously difficult runtime error to diagnose.

# Step 2: Structure Routers Around Spatial Entities

Geospatial APIs frequently suffer from monolithic route files that mix CRUD, spatial analysis, and administrative operations. Clean modeling requires isolating spatial resources by domain boundary: /parcels, /sensors, /routes. Each router owns its Pydantic response models, query builders, and error handlers.

For the full directory convention and dependency injection patterns, see FastAPI Routers for PostGIS Tables.

# routers/parcels.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from schemas.parcel import ParcelRead, ParcelCreate
from db.session import get_db_session

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

@router.get("/{parcel_id}", response_model=ParcelRead)
async def get_parcel(
    parcel_id: int,
    db: AsyncSession = Depends(get_db_session),
):
    result = await db.get(Parcel, parcel_id)
    if result is None:
        raise HTTPException(status_code=404, detail="Parcel not found")
    return result

Keep Pydantic schemas in a parallel schemas/ directory, separating Create, Update, Read, and SpatialQuery variants. Inject spatial validation middleware at the router level to normalize incoming bounding boxes and reject out-of-bounds coordinates before they reach the database.

# Step 3: Implement Async Connection Pooling

Spatial queries are I/O heavy. Without proper connection management, concurrent requests exhaust your database pool and trigger cascading timeouts. FastAPI’s dependency injection system provides a clean mechanism for managing asyncpg connection lifecycles.

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

# Pool size formula: (2 × CPU cores) + effective_spindle_count
# For a 4-core host with SSD: pool_size=9, max_overflow=5
engine = create_async_engine(
    "postgresql+asyncpg://user:pass@localhost/gis_db",
    pool_size=9,
    max_overflow=5,
    pool_timeout=30,
    pool_pre_ping=True,  # detect stale connections before use
)
AsyncSessionFactory = async_sessionmaker(engine, expire_on_commit=False)

async def get_db_session():
    async with AsyncSessionFactory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

Avoid synchronous psycopg2 or blocking ORM calls inside async endpoints — they freeze the event loop and collapse throughput under concurrent load. Use selectinload or explicit JOIN queries instead of lazy loading to eliminate N+1 spatial relationship fetches.

# Step 4: Optimize Payload Serialization

A single complex polygon with thousands of vertices can inflate a JSON response to several megabytes. Modern spatial APIs must support format negotiation and selective serialization. The GeoJSON vs GeoParquet Serialization decision matrix covers the full trade-off between human-readable interchange and columnar compression for analytical workloads.

from fastapi import Request
from fastapi.responses import StreamingResponse, Response
import orjson

@router.get("/export", summary="Stream spatial dataset with format negotiation")
async def export_parcels(
    request: Request,
    db: AsyncSession = Depends(get_db_session),
):
    accept = request.headers.get("Accept", "application/geo+json")

    if "vnd.apache.parquet" in accept:
        # Return GeoParquet for data pipeline consumers
        return await stream_geoparquet(db)

    # Default: stream GeoJSON with coordinate precision trimming
    async def geojson_stream():
        yield b'{"type":"FeatureCollection","features":['
        first = True
        async for row in await db.stream(select(Parcel)):
            feature = row_to_geojson_feature(row, precision=6)
            prefix = b"" if first else b","
            yield prefix + orjson.dumps(feature)
            first = False
        yield b"]}"

    return StreamingResponse(geojson_stream(), media_type="application/geo+json")

Apply coordinate precision trimming (6 decimals for metre-level accuracy, 4 for kilometre-level) before serialization. This reduces payload size by 30–50% without perceptible visual loss in web mapping clients.

# Step 5: Design Spatial-Aware Pagination

Traditional offset-based pagination breaks down with spatial datasets. Sorting by id or created_at ignores geographic proximity and produces inconsistent results across pages when used alongside map-viewport filtering. As detailed in Spatial Pagination & Cursor Strategies, cursor-based pagination with bounding-box boundaries maintains deterministic ordering and respects spatial indexes.

import base64, json
from sqlalchemy import select, and_

@router.get("/", response_model=ParcelPage)
async def list_parcels(
    bbox: str | None = None,          # "minx,miny,maxx,maxy" (WGS84)
    cursor: str | None = None,
    limit: int = 50,
    db: AsyncSession = Depends(get_db_session),
):
    filters = []

    if bbox:
        minx, miny, maxx, maxy = [float(v) for v in bbox.split(",")]
        filters.append(
            Parcel.geom.ST_Intersects(
                f"SRID=4326;POLYGON(({minx} {miny},{maxx} {miny},"
                f"{maxx} {maxy},{minx} {maxy},{minx} {miny}))"
            )
        )

    if cursor:
        last_id = json.loads(base64.b64decode(cursor))["last_id"]
        filters.append(Parcel.id > last_id)

    stmt = (
        select(Parcel)
        .where(and_(*filters))
        .order_by(Parcel.id)
        .limit(limit + 1)  # fetch one extra to detect next page
    )
    rows = (await db.execute(stmt)).scalars().all()

    has_next = len(rows) > limit
    rows = rows[:limit]
    next_cursor = None
    if has_next:
        payload = json.dumps({"last_id": rows[-1].id})
        next_cursor = base64.b64encode(payload.encode()).decode()

    return {"features": rows, "next_cursor": next_cursor}

# Production Code Example

The following route demonstrates the full modeling pattern: geometry validation, async query, ST_AsGeoJSON serialization, and coordinate precision enforcement in a single cohesive endpoint.

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession
from db.session import get_db_session
import orjson

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

@router.get("/{parcel_id}/geojson")
async def get_parcel_geojson(
    parcel_id: int,
    precision: int = Query(default=6, ge=1, le=10),
    db: AsyncSession = Depends(get_db_session),
):
    """
    Return a single parcel as a GeoJSON Feature.
    ST_AsGeoJSON handles coordinate ordering (lon, lat) per RFC 7946.
    precision controls decimal places (6 = ~0.1m accuracy at equator).
    """
    sql = text("""
        SELECT
            id,
            name,
            ST_AsGeoJSON(geom, :precision)::jsonb AS geometry
        FROM parcels
        WHERE id = :parcel_id
    """)
    result = await db.execute(sql, {"parcel_id": parcel_id, "precision": precision})
    row = result.mappings().one_or_none()

    if row is None:
        raise HTTPException(status_code=404, detail=f"Parcel {parcel_id} not found")

    feature = {
        "type": "Feature",
        "id": row["id"],
        "properties": {"name": row["name"]},
        "geometry": row["geometry"],
    }
    return Response(content=orjson.dumps(feature), media_type="application/geo+json")

# Verification & Testing

After implementing the patterns above, confirm correctness with the following checks.

Curl smoke test:

# Should return HTTP 200 with Content-Type: application/geo+json
curl -s -I "http://localhost:8000/parcels/1/geojson" | grep -E "HTTP|content-type"

# Check geometry coordinates are in lon, lat order
curl -s "http://localhost:8000/parcels/1/geojson" | python3 -c \
  "import sys,json; g=json.load(sys.stdin); print(g['geometry']['coordinates'])"

EXPLAIN ANALYZE — confirm GIST index is used:

EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT id, ST_AsGeoJSON(geom, 6)
FROM parcels
WHERE ST_Intersects(
    geom,
    ST_MakeEnvelope(-0.5, 51.3, 0.5, 51.6, 4326)
);
-- Expected: "Index Scan using parcels_geom_idx on parcels"
-- Red flag: "Seq Scan" means the GIST index is not being used

Unit test skeleton:

import pytest
from httpx import AsyncClient
from main import app

@pytest.mark.asyncio
async def test_parcel_geojson_structure():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/parcels/1/geojson")
    assert response.status_code == 200
    body = response.json()
    assert body["type"] == "Feature"
    assert "geometry" in body
    assert body["geometry"]["type"] == "Polygon"
    # Validate coordinate order: first coordinate is [lon, lat] — lon must be in [-180, 180]
    first_coord = body["geometry"]["coordinates"][0][0]
    assert -180 <= first_coord[0] <= 180, "Longitude out of range — likely swapped with latitude"

# Failure Modes & Edge Cases

1. Silent coordinate inversion. Accepting (lat, lon) instead of (lon, lat) creates geometries that map to the wrong hemisphere. ST_IsValid cannot catch this — only a bounds check against the expected region will. Validate incoming coordinate arrays with a Pydantic @field_validator that asserts longitude is in [-180, 180] and latitude in [-90, 90].

2. GIST index not used after type migration. If you add a spatial column to an existing table and then run CREATE INDEX CONCURRENTLY, the planner may still prefer a sequential scan until ANALYZE parcels; is run. Always run ANALYZE after bulk inserts or schema changes.

3. N+1 spatial joins under lazy loading. SQLAlchemy 2.0 async sessions raise MissingGreenlet if lazy relationships are accessed outside the session scope. Use selectinload or write explicit JOIN queries — do not rely on the ORM’s default lazy strategy in async contexts.

4. Pool exhaustion under slow spatial queries. ST_Buffer and ST_Union on large polygon sets can hold a connection for seconds. With the default pool size, 10 simultaneous slow queries will exhaust a pool of 10. Set statement_timeout at the PostgreSQL role level (ALTER ROLE api_user SET statement_timeout = '5s') and handle asyncpg.exceptions.QueryCanceledError gracefully.

5. GEOSException: TopologyException on malformed input. Invalid geometries (self-intersecting rings, unclosed polygons) cause PostGIS to raise this at query time, not at insert time, unless AddGeometryColumn constraints or a CHECK (ST_IsValid(geom)) constraint is in place. Add the validity check constraint during schema creation, not after data has been loaded.

6. CheckConstraint not enforced on bulk loads. COPY and INSERT ... SELECT bypass row-level triggers but do enforce CHECK constraints. However, if you disable constraints for a bulk load (SET session_replication_role = replica), re-enable and validate with SELECT id FROM parcels WHERE NOT ST_IsValid(geom) before re-exposing the API.

# Performance Notes

• GIST vs BRIN indexes: GIST indexes are the default for spatial columns and support all PostGIS operators. BRIN indexes are smaller but only useful for spatially sorted data (e.g., sensor readings inserted in geographic order). For most API use cases, GIST is the correct choice.
• ST_Simplify before serialization: For zoom levels below 10 in web mapping, apply ST_Simplify(geom, 0.001) server-side before serializing. This can reduce geometry vertex counts by 80% with no visible impact at the target zoom.
• Async vs sync latency: On a 4-core server, an async FastAPI endpoint with asyncpg handles ~3× the concurrent spatial requests of a synchronous psycopg2 endpoint before latency degrades, because spatial I/O wait time is spent yielding the event loop rather than blocking it.
• Connection pool tuning: Start with pool_size = (2 × vCPU) + 1. For workloads dominated by long-running analytical queries, reduce pool_size and increase max_overflow to prevent pool exhaustion while keeping idle connections low.

# Frequently Asked Questions

# When should I use PostGIS geometry vs geography types?

Use geography for data spanning large extents where spheroidal accuracy matters (global asset tracking, shipping routes). Use geometry with an explicit SRID for localized, high-throughput applications like municipal zoning or indoor mapping — it is faster and supports a wider set of PostGIS functions including all topology operations.

# How do I prevent N+1 spatial queries in FastAPI?

Use selectinload or explicit JOIN queries with SQLAlchemy 2.0 async sessions. Never rely on lazy loading inside async endpoints — lazy loads trigger synchronous I/O that blocks the event loop and causes cascading timeouts under concurrent load.

# What is the correct coordinate order for GeoJSON in a FastAPI API?

RFC 7946 mandates longitude, then latitude (x, y order). Document this explicitly in your OpenAPI schema and validate it with a Pydantic field_validator to prevent client-side coordinate inversion bugs that produce geometries mirrored across the prime meridian.

# Related

• FastAPI Routers for PostGIS Tables — directory layout, dependency injection, and middleware wiring for spatial routers
• Spatial Pagination & Cursor Strategies — cursor encoding, bounding-box pagination, and index-safe ordering
• GeoJSON vs GeoParquet Serialization — format selection decision matrix and streaming implementation
• API Versioning for GIS Endpoints — versioning strategies that keep spatial contracts stable across client generations

← Back to Core Geospatial API Architecture