"""Service-account token for the pg-wire password when the app runs deployed.
Local/dev: DODIL_TOKEN is used directly. Deployed: DODIL_SERVICE_ACCOUNT_ID/SECRET mint + cache a
short-lived client_credentials access token, refreshed before expiry."""
import os, time, threading
import httpx
_TOKEN_URL = os.environ.get("DODIL_OIDC_TOKEN_URL",
"https://id.dodil.io/realms/dodil/protocol/openid-connect/token")
_lock = threading.Lock()
_cache = {"tok": None, "exp": 0.0}
def get_pg_password() -> str:
sa_id = os.environ.get("DODIL_SERVICE_ACCOUNT_ID")
sa_secret = os.environ.get("DODIL_SERVICE_ACCOUNT_SECRET")
if not (sa_id and sa_secret):
return os.environ["DODIL_TOKEN"] # local/dev
now = time.time()
with _lock:
if _cache["tok"] and now < _cache["exp"] - 60:
return _cache["tok"]
r = httpx.post(_TOKEN_URL, data={"grant_type":"client_credentials",
"client_id":sa_id, "client_secret":sa_secret}, timeout=15.0)
r.raise_for_status()
d = r.json()
_cache["tok"] = d["access_token"]
_cache["exp"] = now + float(d.get("expires_in", 300))
return _cache["tok"]