"""Thin wrapper around the Asana REST API with proxy/retry support."""
from __future__ import annotations
import time
from typing import Any
import httpx
import config
from auth import get_pat
def _build_client() -> httpx.Client:
return httpx.Client(
base_url=config.ASANA_API_BASE,
headers={
"Authorization": f"Bearer {get_pat()}",
"Accept": "application/json",
},
proxy=config.PROXY_URL,
verify=config.CA_BUNDLE,
timeout=httpx.Timeout(
connect=config.CONNECT_TIMEOUT,
read=config.READ_TIMEOUT,
write=config.READ_TIMEOUT,
pool=config.CONNECT_TIMEOUT,
),
)
def _request_with_retry(
method: str,
path: str,
*,
json_body: dict[str, Any] | None = None,
) -> httpx.Response:
"""Execute an HTTP request with exponential backoff on 429/5xx."""
client = _build_client()
last_exc: Exception | None = None
for attempt in range(config.MAX_RETRIES + 1):
try:
resp = client.request(method, path, json=json_body)
if resp.status_code == 429:
retry_after = int(resp.headers.get("Retry-After", "5"))
time.sleep(retry_after)
continue
if resp.status_code >= 500 and attempt < config.MAX_RETRIES:
time.sleep(2 ** attempt)
continue
resp.raise_for_status()
return resp
except httpx.HTTPStatusError:
raise
except httpx.HTTPError as exc:
last_exc = exc
if attempt < config.MAX_RETRIES:
time.sleep(2 ** attempt)
continue
raise
raise last_exc or RuntimeError("Request failed after retries")
def get_task(task_gid: str) -> dict[str, Any]:
resp = _request_with_retry("GET", f"/tasks/{task_gid}")
return resp.json()["data"]
def delete_task(task_gid: str) -> None:
_request_with_retry("DELETE", f"/tasks/{task_gid}")
def get_project(project_gid: str) -> dict[str, Any]:
resp = _request_with_retry("GET", f"/projects/{project_gid}")
return resp.json()["data"]
def delete_project(project_gid: str) -> None:
_request_with_retry("DELETE", f"/projects/{project_gid}")
def get_user_me() -> dict[str, Any]:
resp = _request_with_retry("GET", "/users/me")
return resp.json()["data"]
