f8cd37ac8c
Includes: API (Express/TypeScript), Astro site, Python workers, document generators, FCC compliance tools, Canada CRTC formation, Ansible infrastructure, and deployment scripts. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
137 lines
4.7 KiB
Python
137 lines
4.7 KiB
Python
"""
|
|
Flowroute API client for Canadian DID provisioning.
|
|
|
|
Docs: https://developer.flowroute.com/api/numbers/v2.0/
|
|
Auth: HTTP Basic with access_key:secret_key.
|
|
"""
|
|
import logging
|
|
import os
|
|
from typing import Optional
|
|
|
|
import requests
|
|
|
|
LOG = logging.getLogger("tests.flowroute")
|
|
|
|
API_BASE = "https://api.flowroute.com"
|
|
|
|
|
|
class FlowrouteClient:
|
|
def __init__(
|
|
self,
|
|
access_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
):
|
|
self.access_key = access_key or os.environ["FLOWROUTE_ACCESS_KEY"]
|
|
self.secret_key = secret_key or os.environ["FLOWROUTE_SECRET_KEY"]
|
|
self.session = requests.Session()
|
|
self.session.auth = (self.access_key, self.secret_key)
|
|
self.session.headers["Accept"] = "application/json"
|
|
|
|
# BC area codes in preference order
|
|
BC_AREA_CODES = ["1604", "1778", "1236", "1250"]
|
|
|
|
def search_available_dids(
|
|
self,
|
|
starts_with: str = "",
|
|
limit: int = 5,
|
|
number_type: str = "standard",
|
|
province: str = "BC",
|
|
) -> list[dict]:
|
|
"""Search for available Canadian DIDs.
|
|
|
|
If starts_with is empty and province is BC, tries all BC area codes
|
|
(604, 778, 236, 250) until DIDs are found.
|
|
|
|
Args:
|
|
starts_with: Number prefix (e.g. "1604"). If empty, searches all BC codes.
|
|
limit: Max results per area code
|
|
number_type: 'standard' or 'tollfree'
|
|
province: Province hint — used to pick area codes when starts_with is empty
|
|
|
|
Returns: List of {"did": "+1604...", "monthly_cost": "1.25", ...}
|
|
"""
|
|
prefixes = [starts_with] if starts_with else self.BC_AREA_CODES
|
|
|
|
for prefix in prefixes:
|
|
params = {
|
|
"starts_with": prefix,
|
|
"limit": limit,
|
|
"number_type": number_type,
|
|
}
|
|
|
|
r = self.session.get(f"{API_BASE}/v2/numbers/available", params=params, timeout=15)
|
|
|
|
if r.status_code != 200:
|
|
LOG.warning("Flowroute search %s: HTTP %d", prefix, r.status_code)
|
|
continue
|
|
|
|
data = r.json()
|
|
results = []
|
|
for item in data.get("data", []):
|
|
did_id = item.get("id", "")
|
|
attrs = item.get("attributes", {})
|
|
results.append({
|
|
"did": did_id,
|
|
"rate_center": attrs.get("rate_center", ""),
|
|
"state": attrs.get("state", ""),
|
|
"monthly_cost": attrs.get("monthly_cost", ""),
|
|
"number_type": attrs.get("number_type", ""),
|
|
})
|
|
|
|
if results:
|
|
LOG.info("Flowroute search '%s': %d results", prefix, len(results))
|
|
return results
|
|
|
|
LOG.info("Flowroute search '%s': 0 results, trying next area code", prefix)
|
|
|
|
LOG.warning("Flowroute: no DIDs found in any BC area code")
|
|
return []
|
|
|
|
def purchase_did(self, did: str) -> dict:
|
|
"""Purchase a DID.
|
|
|
|
Args:
|
|
did: The phone number to purchase (e.g. "16045551234")
|
|
|
|
Returns: {"success": bool, "did": str, ...}
|
|
"""
|
|
r = self.session.post(f"{API_BASE}/v2/numbers/{did}", timeout=15)
|
|
|
|
if r.status_code in (200, 201):
|
|
LOG.info("Flowroute purchased DID: %s", did)
|
|
return {"success": True, "did": did, "raw": r.json()}
|
|
else:
|
|
LOG.error("Flowroute purchase failed: %d %s", r.status_code, r.text[:200])
|
|
return {"success": False, "did": did, "error": r.text[:200]}
|
|
|
|
def release_did(self, did: str) -> dict:
|
|
"""Release (cancel) a purchased DID."""
|
|
r = self.session.delete(f"{API_BASE}/v2/numbers/{did}", timeout=15)
|
|
|
|
if r.status_code in (200, 204):
|
|
LOG.info("Flowroute released DID: %s", did)
|
|
return {"success": True, "did": did}
|
|
else:
|
|
LOG.error("Flowroute release failed: %d %s", r.status_code, r.text[:200])
|
|
return {"success": False, "did": did, "error": r.text[:200]}
|
|
|
|
def list_dids(self, limit: int = 10) -> list[dict]:
|
|
"""List currently owned DIDs."""
|
|
r = self.session.get(
|
|
f"{API_BASE}/v2/numbers",
|
|
params={"limit": limit},
|
|
timeout=15,
|
|
)
|
|
if r.status_code != 200:
|
|
return []
|
|
return [
|
|
{"did": item.get("id", ""), **item.get("attributes", {})}
|
|
for item in r.json().get("data", [])
|
|
]
|
|
|
|
def ping(self) -> bool:
|
|
"""Verify API credentials by listing DIDs (simplest authenticated call)."""
|
|
r = self.session.get(f"{API_BASE}/v2/numbers", params={"limit": 1}, timeout=10)
|
|
ok = r.status_code == 200
|
|
LOG.info("Flowroute API ping: %s", "OK" if ok else f"FAIL {r.status_code}")
|
|
return ok
|