Podium Multi Location Router

Podium Multi-Location Router

Overview

Route Podium API calls across multiple physical locations and operate the routing layer in production. This is not a tutorial on Podium's location model — it is the per-location dispatch code your integration runs when Sydney and Burleigh Heads share a single OAuth app, when an agency onboards five new stores in one afternoon, and when a compliance auditor walks in and asks "which location's webchat received that contact write at 14:07 UTC last Tuesday."

The six production failures this skill prevents:

1. Writes to wrong location silently — Sydney's contact write lands on Burleigh Heads' contact list because the calling code resolved location_uid from a stale lookup or hard-coded the wrong UID. The Podium API accepts it, returns 200, and the data sits in the wrong location with no error surface. Naive integrations discover this when a customer reports "I'm getting review requests for a store I've never been to."
2. Credential cross-contamination — Sydney's OAuth token gets reused for a write to Burleigh Heads. The Podium API does not reject this because the OAuth app is org-scoped and both locations live under the same org. The write succeeds; the audit trail attributes it to Sydney's credential; post-incident forensics cannot tell whether the write was authorized by Sydney's operator or by a bug.
3. Audit trail missing location-ID — log lines say wrote contact name=Jane Doe without saying which location_uid received the write. When a compliance question lands months later, the integration cannot answer "which location received this customer's data?" The answer determines GDPR data-subject scope and PCI cardholder-data exposure.
4. Bulk onboarding race condition — onboarding 5 new locations in one operation spawns 5 OAuth authorization flows. One fails partway (a user closed the consent tab, a redirect URI mismatched), the orchestrator continues with the rest, and the credentials map ends up with 4 valid entries + 1 dangling half-record. Subsequent operations against the half-onboarded location either crash or, worse, fall through to a default credential and write to the wrong place.
5. Location-ID verification skipped on write — the application code passes an arbitrary location_uid into the Podium call. Podium accepts any well-formed UID and returns 403 only if the current token genuinely has no access. The 403 is swallowed by an upstack handler ("just a stale auth, will retry"), and the integration silently stops working for one location while the rest keep flowing.
6. Rate-limits not isolated per location — the integration uses a single shared token bucket. Sydney runs a holiday review-request burst at 2pm and burns the org-wide quota; Burleigh Heads' webchats start returning 429 with no explanation visible to the Burleigh Heads operator who has done nothing wrong.

Prerequisites

• Python 3.10+
• podium-auth skill installed; this skill consumes its PodiumAuth and PodiumOrgRouter classes
• podium-rate-limit-survival skill installed; this skill consumes its per-bucket budget primitive
• Podium OAuth app with locations.read scope granted (verifying a location_uid requires GET /v4/locations)
• A writable directory for the audit log (default ./audit-log/podium-router.jsonl)
• A persistent location-credential map (file, AWS Secrets Manager, GCP Secret Manager, or SOPS-encrypted YAML)

Instructions

Build in this order. Each section neutralizes one production failure mode.

1. Per-location credential isolation (neutralizes cross-contamination)

The router holds one PodiumAuth instance per location_uid, never per org. Two locations under the same org get two separate auth instances because the audit trail must attribute every call to a specific location and a specific credential record — even if the underlying OAuth refresh token happens to be shared at the org level.

import asyncio, time
from dataclasses import dataclass, field
from typing import Optional
from podium_auth import PodiumAuth, PodiumOrgRouter, PodiumAuthError

@dataclass
class LocationCredential:
    location_uid: str
    org_slug: str
    client_id: str
    client_secret: str
    refresh_token_file: str
    verified_at: float = 0.0   # unix seconds; 0 = never verified

class LocationRouter:
    def __init__(self, credentials: list[LocationCredential], audit_log_path: str):
        self._creds: dict[str, LocationCredential] = {c.location_uid: c for c in credentials}
        self._auths: dict[str, PodiumAuth] = {}
        self._audit_log_path = audit_log_path
        self._verify_lock = asyncio.Lock()

    def get_client(self, location_uid: str) -> "PodiumLocationClient":
        cred = self._creds.get(location_uid)
        if not cred:
            raise UnknownLocationError(location_uid)
        if location_uid not in self._auths:
            self._auths[location_uid] = PodiumAuth(
                client_id=cred.client_id,
                client_secret=cred.client_secret,
                refresh_token=load_refresh_token(cred.refresh_token_file),
            )
        return PodiumLocationClient(
            location_uid=location_uid,
            auth=self._auths[location_uid],
            router=self,
        )

The map key is location_uid, never org_slug. Two locations under the same org get two map entries with the same client_id/client_secret but distinct refresh_token_file paths. This is the line that prevents Sydney's token from being reused for a Burleigh Heads write — the router has no syntactic path from a location_uid to a different location's auth.

2. Pre-flight location-ID verification (neutralizes silent wrong-location writes)

Before any write, the router confirms the location_uid exists in the set returned by GET /v4/locations for the current credential. Cache the verification with a short TTL — typically 1 hour — so a re-org of locations on the Podium side eventually propagates without re-verifying every call.

VERIFY_TTL_SECONDS = 3600

async def ensure_location_verified(self, location_uid: str) -> None:
    cred = self._creds[location_uid]
    if time.time() - cred.verified_at < VERIFY_TTL_SECONDS:
        return
    async with self._verify_lock:
        if time.time() - cred.verified_at < VERIFY_TTL_SECONDS:
            return
        auth = self._auths[location_uid]
        token = await auth.get_token()
        async with httpx.AsyncClient(timeout=10) as c:
            r = await c.get(
                "https://api.podium.com/v4/locations",
                headers={"Authorization": f"Bearer {token}"},
            )
        if r.status_code != 200:
            raise LocationVerificationError(location_uid, r.status_code, r.text)
        ids = {loc["uid"] for loc in r.json().get("locations", [])}
        if location_uid not in ids:
            raise LocationNotInScopeError(location_uid, sorted(ids))
        cred.verified_at = time.time()

LocationNotInScopeError is the loud version of "Podium returned 403 silently." It fires before the call is made, so the caller cannot accidentally write to a location their token does not own. The 1-hour TTL keeps the verification cost negligible while bounding the staleness window.

3. Structured audit log on every call (neutralizes audit-trail gaps)

Every API call routed through this skill emits one JSON line to the audit log. The fields are fixed: {ts, location_uid, org_slug, endpoint, method, status, request_id, latency_ms}. No tokens, no request bodies, no PII — just the routing fingerprint.

import json, os, time, uuid

def emit_audit(self, *, location_uid: str, endpoint: str, method: str,
               status: int, latency_ms: float, request_id: str) -> None:
    cred = self._creds[location_uid]
    record = {
        "ts": time.time(),
        "location_uid": location_uid,
        "org_slug": cred.org_slug,
        "endpoint": endpoint,
        "method": method,
        "status": status,
        "request_id": request_id,
        "latency_ms": round(latency_ms, 2),
    }
    # Append-only; one line per call. Operators rotate with logrotate, not in-process.
    os.makedirs(os.path.dirname(self._audit_log_path), exist_ok=True)
    with open(self._audit_log_path, "a") as f:
        f.write(json.dumps(record) + "\n")

The request_id is generated by the client wrapper (uuid.uuid4().hex) and also returned in any error so the operator can correlate a customer complaint to a single audit-log line. Six months later, when compliance asks "which location received this write?", audit_log_query.py answers in one second.