Server Integration — Python / FastAPI

Install dependencies

pip install fastapi uvicorn pickpoint websockets python-dotenv

Environment & app setup

# .env
PICKPOINT_KEY=your_key_here

# main.py
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI
from pickpoint import Client, Config
from dotenv import load_dotenv

load_dotenv()

pp: Client


@asynccontextmanager
async def lifespan(_app: FastAPI):
    global pp
    async with Client(Config(api_key=os.environ["PICKPOINT_KEY"])) as client:
        pp = client
        yield


app = FastAPI(lifespan=lifespan)

Geocoding endpoints

# ── Geocoding ─────────────────────────────────────────────────────────────

@app.get("/geo/forward")
async def forward(q: str, limit: int = 5, countrycodes: str | None = None,
                  viewbox: str | None = None, bounded: int | None = None,
                  language: str | None = None):
    params = {"q": q, "limit": str(limit)}
    if countrycodes: params["countrycodes"] = countrycodes
    if viewbox:      params["viewbox"] = viewbox
    if bounded:      params["bounded"] = str(bounded)
    if language:     params["language"] = language
    return await pp.forward(params)


@app.get("/geo/reverse")
async def reverse(lat: float, lon: float, zoom: int = 18, language: str | None = None):
    params = {"lat": str(lat), "lon": str(lon), "zoom": str(zoom)}
    if language: params["language"] = language
    return await pp.reverse(params)


@app.get("/geo/search")
async def search(q: str, limit: int = 6, viewbox: str | None = None,
                 countrycodes: str | None = None):
    params = {"q": q, "limit": str(limit)}
    if viewbox:      params["viewbox"] = viewbox
    if countrycodes: params["countrycodes"] = countrycodes
    return await pp.search(params)

Routing endpoints

# ── Routing ───────────────────────────────────────────────────────────────
from pydantic import BaseModel
from typing import List

class Location(BaseModel):
    lat: float
    lon: float

class RouteRequest(BaseModel):
    locations: List[Location]
    costing: str = "auto"
    costing_options: dict = {}
    units: str = "km"

@app.post("/route")
async def route(body: RouteRequest):
    return await pp.route(body.model_dump())


@app.post("/route/optimized")
async def route_optimized(body: RouteRequest):
    return await pp.optimized_route(body.model_dump())

Device tracking - REST

# ── Device tracking - REST ────────────────────────────────────────────────

@app.get("/devices")
async def list_devices():
    return await pp.devices.list()

Track history is REST-only today — see the Device Tracking guide.

Device tracking - WebSocket relay

# ── Device tracking - WebSocket subscriber relay ─────────────────────────
# Run separately:  python ws_relay.py
import asyncio
import json
import os
from pickpoint import Config, mint_client_tokens
from pickpoint.tracking import Config as TrackingConfig, ListenerAuth, connect
import websockets

KEY = os.environ["PICKPOINT_KEY"]

async def relay(device_uid: str, client_ws):
    pair = await mint_client_tokens(
        Config(api_key=KEY),
        scopes=["devices"],
        ttl_sec=600,
    )
    session = await connect(
        TrackingConfig(
            endpoint="wss://tracking.pickpoint.io",
            listener=ListenerAuth(access_token=pair.access_token),
            subscribe=[device_uid],
        )
    )
    async for msg in session:
        if msg.loc:
            await client_ws.send(json.dumps({
                "latitude":  msg.loc.point.latitude,
                "longitude": msg.loc.point.longitude,
            }))

async def handler(ws):
    qs = ws.path.split("?", 1)[-1] if "?" in ws.path else ""
    params = dict(p.split("=", 1) for p in qs.split("&") if "=" in p)
    uid = params.get("deviceUid")
    if not uid:
        await ws.close(code=1008, reason="deviceUid required")
        return
    await relay(uid, ws)

asyncio.run(websockets.serve(handler, "0.0.0.0", 3001))

Run with python ws_relay.py alongside the FastAPI app. Start the API with uvicorn main:app --reload.