Backend Integration - Python / FastAPI

Install dependencies

pip install fastapi uvicorn httpx python-socketio websockets python-dotenv

Environment & app setup

# .env
PICKPOINT_KEY=your_key_here

# main.py
import os
from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import JSONResponse
import httpx
from dotenv import load_dotenv

load_dotenv()
app  = FastAPI()
KEY  = os.environ["PICKPOINT_KEY"]
BASE = "https://api.pickpoint.io/v2"
HDR  = {"X-Api-Key": KEY}

Geocoding endpoints

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

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

    async with httpx.AsyncClient() as client:
        r = await client.get(f"{BASE}/geocode/forward", params=params, headers=HDR)
    return JSONResponse(r.json(), status_code=r.status_code)


@app.get("/geo/reverse")
async def reverse(lat: float, lon: float, zoom: int = 18, language: str = None):
    params = {"lat": lat, "lon": lon, "zoom": zoom}
    if language: params["language"] = language
    async with httpx.AsyncClient() as client:
        r = await client.get(f"{BASE}/geocode/reverse", params=params, headers=HDR)
    return JSONResponse(r.json(), status_code=r.status_code)


@app.get("/geo/search")
async def search(q: str, limit: int = 6, viewbox: str = None, countrycodes: str = None):
    params = {"q": q, "limit": limit}
    if viewbox:      params["viewbox"]      = viewbox
    if countrycodes: params["countrycodes"] = countrycodes
    async with httpx.AsyncClient() as client:
        r = await client.get(f"{BASE}/address/search", params=params, headers=HDR)
    return JSONResponse(r.json(), status_code=r.status_code)

Routing endpoints

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

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):
    async with httpx.AsyncClient() as client:
        r = await client.post(
            f"{BASE}/route",
            json=body.model_dump(),
            headers={**HDR, "Content-Type": "application/json"},
        )
    return JSONResponse(r.json(), status_code=r.status_code)


@app.post("/route/optimized")
async def route_optimized(body: RouteRequest):
    async with httpx.AsyncClient() as client:
        r = await client.post(
            f"{BASE}/route/optimized",
            json=body.model_dump(),
            headers={**HDR, "Content-Type": "application/json"},
        )
    return JSONResponse(r.json(), status_code=r.status_code)

Device tracking - REST

# ── Device Tracking - REST ────────────────────────────────────────────────

@app.get("/devices")
async def list_devices():
    async with httpx.AsyncClient() as client:
        r = await client.get(f"{BASE}/devices", headers=HDR)
    return JSONResponse(r.json(), status_code=r.status_code)


@app.get("/devices/{uid}/tracks")
async def device_tracks(uid: str, from_: str = Query(None, alias="from"),
                        to: str = None):
    params = {}
    if from_: params["from"] = from_
    if to:    params["to"]   = to
    async with httpx.AsyncClient() as client:
        r = await client.get(f"{BASE}/devices/{uid}/tracks", params=params, headers=HDR)
    return JSONResponse(r.json(), status_code=r.status_code)

Device tracking - WebSocket relay

# ── Device Tracking - WebSocket subscriber relay ─────────────────────────
# Run separately:  python ws_relay.py
import asyncio
import websockets
import socketio
import json

sio = socketio.AsyncClient()

async def relay(device_uid: str, client_ws):
    """Subscribe to PickPoint and forward events to the browser client."""
    await sio.connect(
        "wss://ws.pickpoint.io",
        transports=["websocket"],
        socketio_path="/v2/tracking",
        auth={"x-api-key": KEY},
    )
    await sio.emit("device:subscribe", {"deviceUid": device_uid})

    @sio.on("location:added")
    async def on_location(data):
        await client_ws.send(json.dumps(data))

    await sio.wait()

async def handler(ws, path):
    # ws://your-server/tracking-relay?deviceUid=d_xxx
    import urllib.parse
    qs = urllib.parse.parse_qs(urllib.parse.urlparse(path).query)
    uid = qs.get("deviceUid", [None])[0]
    if not uid:
        await ws.close(1008, "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.