Route Podium API calls across multiple physical locations with strict per-location credential isolation, pre-flight location-ID verification, an immutable audit trail of every write, idempotent bulk onboarding, and per-location rate-limit budgets that cannot starve each other. Use when running Podium for more than one physical store (an agency operator managing 50+ accounts, a multi-store SMB with 2+ locations, or a compliance team that needs to prove which location received which write). Trigger with "podium multi-location", "podium location router", "podium per-location", "podium location audit", "podium bulk onboarding", "podium location_uid verification".
Instalar com Codex ou Claude Copie este prompt, cole no Codex, Claude ou outro assistente e deixe que ele revise a página da skill e instale para você.
Um comando direto ignora o prompt de revisão. Verifique a origem antes de executá-lo.
Instruções da origem · Visualização somente leitura
name
podium-multi-location-router
description
Route Podium API calls across multiple physical locations with strict per-location credential isolation, pre-flight location-ID verification, an immutable audit trail of every write, idempotent bulk onboarding, and per-location rate-limit budgets that cannot starve each other. Use when running Podium for more than one physical store (an agency operator managing 50+ accounts, a multi-store SMB with 2+ locations, or a compliance team that needs to prove which location received which write). Trigger with "podium multi-location", "podium location router", "podium per-location", "podium location audit", "podium bulk onboarding", "podium location_uid verification".
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:
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."
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.
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.
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.
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.
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.
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 importOptionalfrom podium_auth import PodiumAuth, PodiumOrgRouter, PodiumAuthError
@dataclassclassLocationCredential:
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 verifiedclassLocationRouter:
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()
defget_client(self, location_uid: str) -> "PodiumLocationClient":
cred = self._creds.get(location_uid)
ifnot cred:
raise UnknownLocationError(location_uid)
if location_uid notinself._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.
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 = 3600asyncdefensure_location_verified(self, location_uid: str) -> None:
cred = self._creds[location_uid]
if time.time() - cred.verified_at < VERIFY_TTL_SECONDS:
returnasyncwithself._verify_lock:
if time.time() - cred.verified_at < VERIFY_TTL_SECONDS:
return
auth = self._auths[location_uid]
token = await auth.get_token()
asyncwith 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 notin 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
defemit_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)
withopen(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.
Onboarding multiple locations in one operation must be resumable. The orchestrator processes one location at a time, writes each location's credential record to the map atomically, and is safe to re-run after partial failure — already-onboarded locations are detected and skipped.
@dataclassclassOnboardingResult:
location_uid: str
status: str# "onboarded" | "skipped_existing" | "failed"
error: Optional[str] = Noneasyncdefonboard_locations(self, new: list[LocationCredential]) -> list[OnboardingResult]:
results: list[OnboardingResult] = []
for cred in new:
if cred.location_uid inself._creds:
results.append(OnboardingResult(cred.location_uid, "skipped_existing"))
continuetry:
# 1. Persist credential record atomically BEFORE adding to in-memory mapself._persist_credential(cred)
# 2. Build auth + verify location_uid is reachableself._auths[cred.location_uid] = PodiumAuth(
client_id=cred.client_id,
client_secret=cred.client_secret,
refresh_token=load_refresh_token(cred.refresh_token_file),
)
self._creds[cred.location_uid] = cred
awaitself.ensure_location_verified(cred.location_uid)
results.append(OnboardingResult(cred.location_uid, "onboarded"))
except Exception as e:
# Roll back the half-written entry so a re-run doesn't see a phantom locationself._creds.pop(cred.location_uid, None)
self._auths.pop(cred.location_uid, None)
results.append(OnboardingResult(cred.location_uid, "failed", str(e)))
return results
The critical invariant: a location is either fully onboarded (credential persisted + auth instantiated + verification passed) or fully absent. There is no third state. A re-run after partial failure picks up exactly where the previous run left off.
This skill stacks on top of podium-rate-limit-survival. Each location gets its own token bucket; one location's burst cannot starve another. The default budget per location is conservative — read it from config/settings.yaml and tune per-deployment.
from podium_rate_limit import TokenBucket # from podium-rate-limit-survivalclassPodiumLocationClient:
def__init__(self, location_uid: str, auth: PodiumAuth, router: LocationRouter):
self._location_uid = location_uid
self._auth = auth
self._router = router
# One bucket per location — sized to the per-location plan, NOT the org-wide quota.self._bucket = router.bucket_for(location_uid)
asyncdefcall(self, method: str, path: str, **kwargs) -> httpx.Response:
awaitself._router.ensure_location_verified(self._location_uid)
awaitself._bucket.acquire()
request_id = uuid.uuid4().hex
token = awaitself._auth.get_token()
start = time.monotonic()
asyncwith httpx.AsyncClient(timeout=10) as c:
r = await c.request(
method,
f"https://api.podium.com{path}",
headers={
"Authorization": f"Bearer {token}",
"X-Request-ID": request_id,
},
**kwargs,
)
self._router.emit_audit(
location_uid=self._location_uid,
endpoint=path,
method=method,
status=r.status_code,
latency_ms=(time.monotonic() - start) * 1000,
request_id=request_id,
)
return r
A burst from Sydney drains Sydney's bucket and only Sydney's bucket. Burleigh Heads has its own bucket with its own refill timer and remains operational regardless.
Asking for a client for an unknown location_uid raises immediately. There is no default credential, no "best guess" routing, no fallback to the first registered location. This is the single line that prevents "5 fails, the 6th call falls through to wherever" from being a viable failure mode.
classUnknownLocationError(KeyError):
def__init__(self, location_uid: str):
super().__init__(
f"No credentials for location_uid={location_uid}. "f"Run onboard_location.py to register, then reload the router."
)
classLocationNotInScopeError(PodiumAuthError):
def__init__(self, location_uid: str, scope: list[str]):
super().__init__(
403,
f"location_uid={location_uid} not in /v4/locations scope. "f"Token sees {len(scope)} locations; this UID is not one of them. "f"Verify with verify_location.py."
)
Error Handling
HTTP Status
Error
Root Cause
Action
403 Forbidden
LocationNotInScopeError
location_uid not in token's /v4/locations set
Run verify_location.py; confirm the credential is for the right org
404 Not Found
LocationDeletedError
Location was deleted on the Podium side
Remove from credentials map; alert the operator
N/A
UnknownLocationError
Caller asked for a location_uid not in the map
Re-run onboard_location.py for the missing UID
N/A
OnboardingPartialFailure
Bulk onboarding completed N-1 of N locations
Re-run; idempotent — already-onboarded are skipped
429 Too Many Requests
LocationRateLimitExceeded
Per-location bucket exhausted
Bucket auto-refills; respect Retry-After; tune bucket in config/settings.yaml
401 Unauthorized
underlying PodiumAuthError
Token refresh failed for this specific location
Surface to ops; auth-side recovery is podium-auth's domain
