Building a Geospatial REST API with FastAPI and PostGIS

Geospatial APIs have a specific set of requirements that general REST API tutorials tend to skip: geometry serialisation, spatial query parameters (radius searches, bounding boxes), GeoJSON response formats, and efficient use of PostGIS spatial indexes. This post builds a complete working example from schema to endpoint.

The stack is FastAPI for the web framework, PostgreSQL + PostGIS for the spatial database, GeoAlchemy2 for the SQLAlchemy integration, and asyncpg for async database access. This is a common production configuration at companies building location-based services.

Project Setup

pip install fastapi uvicorn sqlalchemy geoalchemy2 asyncpg psycopg2-binary shapely

Directory structure:

api/
├── main.py          # FastAPI app and routes
├── database.py      # Async engine and session
├── models.py        # SQLAlchemy ORM models
├── schemas.py       # Pydantic response schemas
└── queries.py       # Spatial query helpers

Database Schema with PostGIS

First, set up PostGIS and create the table. In SQL:

CREATE EXTENSION IF NOT EXISTS postgis;

CREATE TABLE places (
    id          SERIAL PRIMARY KEY,
    name        TEXT NOT NULL,
    category    TEXT NOT NULL,
    description TEXT,
    rating      REAL DEFAULT 0.0,
    geom        GEOMETRY(POINT, 4326) NOT NULL,  -- WGS84 lat/lng
    created_at  TIMESTAMPTZ DEFAULT NOW()
);

-- Spatial index — critical for ST_DWithin performance
CREATE INDEX idx_places_geom ON places USING GIST (geom);

-- Attribute index for category filtering
CREATE INDEX idx_places_category ON places (category);

The GIST index on geom is what makes radius searches fast. Without it, PostGIS does a sequential scan of every row to evaluate ST_DWithin — fine for thousands of records, catastrophic for millions.

SQLAlchemy Model with GeoAlchemy2

# models.py
from sqlalchemy import Column, Integer, String, Float, DateTime, text
from sqlalchemy.orm import DeclarativeBase
from geoalchemy2 import Geometry
from geoalchemy2.shape import to_shape
from shapely.geometry import mapping

class Base(DeclarativeBase):
    pass

class Place(Base):
    __tablename__ = "places"

    id          = Column(Integer, primary_key=True)
    name        = Column(String, nullable=False)
    category    = Column(String, nullable=False)
    description = Column(String)
    rating      = Column(Float, default=0.0)
    geom        = Column(Geometry(geometry_type="POINT", srid=4326), nullable=False)
    created_at  = Column(DateTime(timezone=True), server_default=text("NOW()"))

    @property
    def coordinates(self) -> dict:
        """Return GeoJSON coordinates from PostGIS geometry."""
        shape = to_shape(self.geom)
        return {"lon": shape.x, "lat": shape.y}

Async Database Session

# database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker

DATABASE_URL = "postgresql+asyncpg://user:password@localhost:5432/spatial_db"

engine = create_async_engine(DATABASE_URL, echo=False)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

Pydantic Response Schemas

GeoJSON is the standard format for geospatial API responses. Define Pydantic models that serialise to valid GeoJSON:

# schemas.py
from pydantic import BaseModel, field_validator
from typing import Literal
from datetime import datetime

class PointGeometry(BaseModel):
    type: Literal["Point"] = "Point"
    coordinates: list[float]  # [lon, lat]

class PlaceProperties(BaseModel):
    id: int
    name: str
    category: str
    description: str | None
    rating: float
    created_at: datetime

class PlaceFeature(BaseModel):
    """GeoJSON Feature representing a single place."""
    type: Literal["Feature"] = "Feature"
    geometry: PointGeometry
    properties: PlaceProperties

class FeatureCollection(BaseModel):
    """GeoJSON FeatureCollection."""
    type: Literal["FeatureCollection"] = "FeatureCollection"
    features: list[PlaceFeature]
    count: int

class PlaceCreate(BaseModel):
    name: str
    category: str
    description: str | None = None
    rating: float = 0.0
    lat: float
    lon: float

The API Endpoints

# main.py
from fastapi import FastAPI, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, text
from geoalchemy2.functions import ST_DWithin, ST_MakePoint, ST_Distance
from geoalchemy2 import WKTElement

from database import get_db
from models import Place
from schemas import FeatureCollection, PlaceFeature, PointGeometry, PlaceProperties, PlaceCreate

app = FastAPI(
    title="Geospatial Places API",
    description="A spatial REST API for finding and managing places",
    version="1.0.0"
)

def place_to_feature(place: Place) -> PlaceFeature:
    coords = place.coordinates
    return PlaceFeature(
        geometry=PointGeometry(coordinates=[coords["lon"], coords["lat"]]),
        properties=PlaceProperties(
            id=place.id,
            name=place.name,
            category=place.category,
            description=place.description,
            rating=place.rating,
            created_at=place.created_at
        )
    )

@app.get("/places", response_model=FeatureCollection)
async def list_places(
    lat: float = Query(..., ge=-90, le=90, description="Centre latitude"),
    lon: float = Query(..., ge=-180, le=180, description="Centre longitude"),
    radius_m: float = Query(1000, ge=1, le=50000, description="Search radius in metres"),
    category: str | None = Query(None, description="Filter by category"),
    limit: int = Query(20, ge=1, le=100),
    db: AsyncSession = Depends(get_db)
):
    """
    Find places within a given radius of a coordinate.
    Returns a GeoJSON FeatureCollection ordered by distance.
    """
    centre = ST_MakePoint(lon, lat)

    query = (
        select(Place)
        .where(
            ST_DWithin(
                Place.geom,
                centre.cast(Place.geom.type),
                radius_m / 111_320  # Convert metres to degrees (approximate)
            )
        )
        .order_by(
            ST_Distance(Place.geom, centre.cast(Place.geom.type))
        )
        .limit(limit)
    )

    if category:
        query = query.where(Place.category == category)

    result = await db.execute(query)
    places = result.scalars().all()

    features = [place_to_feature(p) for p in places]
    return FeatureCollection(features=features, count=len(features))

A Note on Metres vs Degrees

The ST_DWithin function with EPSG:4326 geometries takes a distance in degrees, not metres. The conversion above (/ 111_320) is an approximation that works reasonably well at mid-latitudes. For precise metre-based queries, either:

• Cast to a metric CRS: ST_Transform(geom, 27700) before the comparison
• Use ST_DWithin with use_spheroid=true and a GEOGRAPHY column type (more accurate, slightly slower)

# More accurate approach using ST_Transform to British National Grid (metres)
query = (
    select(Place)
    .where(
        ST_DWithin(
            text("ST_Transform(geom, 27700)"),
            text(f"ST_Transform(ST_SetSRID(ST_MakePoint({lon}, {lat}), 4326), 27700)"),
            radius_m
        )
    )
)

Creating Places

from geoalchemy2 import WKTElement

@app.post("/places", response_model=PlaceFeature, status_code=201)
async def create_place(
    payload: PlaceCreate,
    db: AsyncSession = Depends(get_db)
):
    """Add a new place with a geographic coordinate."""
    geom = WKTElement(f"POINT({payload.lon} {payload.lat})", srid=4326)

    place = Place(
        name=payload.name,
        category=payload.category,
        description=payload.description,
        rating=payload.rating,
        geom=geom
    )
    db.add(place)
    await db.commit()
    await db.refresh(place)
    return place_to_feature(place)

@app.get("/places/{place_id}", response_model=PlaceFeature)
async def get_place(place_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(Place).where(Place.id == place_id))
    place = result.scalar_one_or_none()
    if not place:
        raise HTTPException(status_code=404, detail="Place not found")
    return place_to_feature(place)

Running the API

uvicorn api.main:app --reload --port 8000

The automatic OpenAPI documentation is available at http://localhost:8000/docs — including a try-it-out interface for every endpoint.

Test the radius search:

curl "http://localhost:8000/places?lat=52.954&lon=-1.158&radius_m=500&category=cafe"

Response:

{
  "type": "FeatureCollection",
  "features": [
    {
      "type": "Feature",
      "geometry": { "type": "Point", "coordinates": [-1.1545, 52.9532] },
      "properties": {
        "id": 42,
        "name": "Corner Coffee",
        "category": "cafe",
        "rating": 4.7,
        ...
      }
    }
  ],
  "count": 1
}

Production Considerations

A few things to add before shipping this to production:

Connection pooling: asyncpg has its own connection pool; configure it in the engine to match your Postgres max_connections:

engine = create_async_engine(url, pool_size=10, max_overflow=20)

Spatial index coverage: If you filter by category frequently, a composite index (category, geom) using GIST may outperform two separate indexes for filtered radius searches.

Bounding box endpoints: For map viewport queries, add a GET /places/bbox endpoint that takes min_lat, min_lon, max_lat, max_lon and uses ST_MakeEnvelope — it avoids the degree/metre conversion issue entirely and is the correct pattern for slippy map tile requests.

Rate limiting: geospatial queries can be expensive; add slowapi or a reverse proxy rate limit to protect the database.

This pattern — FastAPI + PostGIS + GeoAlchemy2 + GeoJSON responses — is the starting point for the spatial data layer of most modern location-based web applications.