Skip to main content

Best Practices

Production patterns for building reliable integrations with the SpatialFlow API.

Pagination​

SpatialFlow uses offset-based pagination with limit and offset query parameters. The default page size is 20 and the maximum is 100.

Example request:

curl "https://api.spatialflow.io/api/v1/geofences?limit=20&offset=0" \
  -H "Authorization: Bearer YOUR_API_KEY"

Response envelope:

{
  "items": [
    { "id": "gf_abc123", "name": "Warehouse Zone", "... ": "..." }
  ],
  "count": 150
}

The count field contains the total number of matching resources. Use it to determine how many pages remain.

Paging through all results in Python:

import requests

API_URL = "https://api.spatialflow.io/api/v1/geofences"
HEADERS = {"Authorization": "Bearer YOUR_API_KEY"}
PAGE_SIZE = 50

def get_all_geofences():
    """Fetch all geofences using offset-based pagination."""
    all_items = []
    offset = 0

    while True:
        response = requests.get(
            API_URL,
            headers=HEADERS,
            params={"limit": PAGE_SIZE, "offset": offset},
        )
        response.raise_for_status()
        data = response.json()

        all_items.extend(data["items"])

        if offset + PAGE_SIZE >= data["count"]:
            break
        offset += PAGE_SIZE

    return all_items
Use Smaller Page Sizes

Request 20-50 items per page for faster responses. Larger pages (100) increase response time and memory usage on both client and server.

Error Handling​

All SpatialFlow API errors return a consistent JSON format with a detail field:

{
  "detail": "Human-readable error message",
  "error_code": "MACHINE_READABLE_CODE",
  "details": { "field": ["Validation error message"] }
}
  • detail (always present) -- Human-readable description. For validation errors, this is an array of error objects instead of a string.
  • error_code (optional) -- Machine-readable identifier for programmatic handling.
  • details (optional) -- Additional context such as field-level validation errors.

TypeScript error handling example:

import axios, { AxiosError } from "axios";

interface ApiError {
  detail: string | Array<{ type: string; loc: string[]; msg: string }>;
  error_code?: string;
  details?: Record<string, string[]>;
}

async function createGeofence(data: GeofenceInput): Promise<Geofence> {
  try {
    const response = await axios.post("/api/v1/geofences", data, {
      headers: { Authorization: `Bearer ${apiKey}` },
    });
    return response.data;
  } catch (err) {
    const error = err as AxiosError<ApiError>;
    const body = error.response?.data;

    if (!body) throw err;

    // Validation errors return detail as an array
    if (Array.isArray(body.detail)) {
      const messages = body.detail.map((e) => `${e.loc.join(".")}: ${e.msg}`);
      throw new Error(`Validation failed: ${messages.join(", ")}`);
    }

    // Check machine-readable error code for programmatic handling
    switch (body.error_code) {
      case "RATE_LIMIT_EXCEEDED":
        // Retry after delay (see Rate Limiting page)
        break;
      case "DUPLICATE_RESOURCE":
        // Handle duplicate name/identifier
        break;
      default:
        throw new Error(body.detail);
    }
  }
}

See the Error Reference for the complete list of error codes and HTTP status patterns.

Webhook Reliability​

SpatialFlow delivers webhooks with automatic retries using exponential backoff. Follow these patterns for reliable consumption.

Respond Quickly​

Your webhook endpoint must return a 2xx status code within 30 seconds. If the request times out or returns a non-2xx status, SpatialFlow will retry delivery.

Webhook Timeout

Process webhooks asynchronously. If your handler takes longer than 30 seconds, acknowledge receipt immediately with a 200 response and process the event in a background job.

Handle Duplicates with Idempotency​

Network retries can cause the same event to be delivered more than once. Always check the event_id to avoid processing duplicates:

from flask import Flask, request, jsonify
import hashlib
import hmac

app = Flask(__name__)
WEBHOOK_SECRET = "your-webhook-secret"
processed_events = set()  # Use a database in production

@app.route("/webhooks/spatialflow", methods=["POST"])
def handle_webhook():
    payload = request.get_data()
    event = request.get_json()

    # 1. Verify signature
    signature = request.headers.get("X-SF-Signature", "")
    if signature.startswith("sha256="):
        signature = signature[7:]
    expected = hmac.new(
        WEBHOOK_SECRET.encode(), payload, hashlib.sha256
    ).hexdigest()
    if not hmac.compare_digest(signature, expected):
        return jsonify({"error": "Invalid signature"}), 401

    # 2. Check idempotency -- skip if already processed
    event_id = event.get("event_id")
    if event_id in processed_events:
        return jsonify({"status": "already_processed"}), 200

    # 3. Process the event
    handle_event(event)
    processed_events.add(event_id)

    return jsonify({"status": "ok"}), 200

Retry Behavior​

SpatialFlow retries failed webhook deliveries using exponential backoff:

AttemptDelay After Failure
115 minutes
230 minutes
31 hour
44 hours
54 hours
64 hours
7(final attempt)

After 7 failed attempts, the webhook is moved to the Dead Letter Queue (DLQ). Check your webhook delivery logs in the Dashboard for troubleshooting. See Webhooks for the full retry timeline.

API Key Management​

Use Environment Variables​

Never hardcode API keys in source code. Store them in environment variables:

# Set the API key as an environment variable
export SPATIALFLOW_API_KEY="sf_live_abc123..."
import os
import requests

api_key = os.environ["SPATIALFLOW_API_KEY"]
response = requests.get(
    "https://api.spatialflow.io/api/v1/geofences",
    headers={"Authorization": f"Bearer {api_key}"},
)

Key Hygiene​

  • Separate keys per environment -- Use different keys for development, staging, and production.
  • Rotate regularly -- Generate new keys periodically and revoke old ones.
  • Revoke compromised keys immediately -- If a key is exposed, revoke it in the Dashboard and generate a replacement.
  • Use scoped permissions -- Create keys with only the permissions they need (e.g., geofences:read for a read-only integration).
Least Privilege

Create API keys with the minimum permissions required. A monitoring dashboard only needs read access, while a device ingestion service needs write access to devices but not to workflows.

Connection Best Practices​

Use Connection Pooling​

Reuse HTTP connections to reduce latency and resource usage:

import requests

# Create a session for connection reuse
session = requests.Session()
session.headers.update({"Authorization": "Bearer YOUR_API_KEY"})

# All requests through the session reuse the underlying TCP connection
geofences = session.get("https://api.spatialflow.io/api/v1/geofences")
devices = session.get("https://api.spatialflow.io/api/v1/devices")

Set Reasonable Timeouts​

Always set explicit timeouts to avoid hanging requests:

  • Connect timeout: 10 seconds
  • Read timeout: 30 seconds
response = session.get(
    "https://api.spatialflow.io/api/v1/geofences",
    timeout=(10, 30),  # (connect, read) in seconds
)

Cache JWT Tokens​

JWT access tokens are valid for 24 hours by default (configurable by workspace admins). Cache them until expiration to avoid unnecessary authentication requests. Only refresh when the token expires or a 401 response is received.

Further Reading​