unofficial maimai DX NET client built on ATProto
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201from __future__ import annotations
import jsonimport osimport socketimport sysimport timeimport urllib.errorimport urllib.parseimport urllib.requestfrom collections.abc import Callable
def request_timeout(env_name: str = "ATP_REQUEST_TIMEOUT", default: str = "30") -> float: return float(os.environ.get(env_name, os.environ.get("REQUEST_TIMEOUT", default)))
def env_int(name: str, default: int) -> int: return max(1, int(os.environ.get(name, str(default))))
def error_summary(error: BaseException) -> str: if isinstance(error, urllib.error.HTTPError): body = "" try: body = error.read().decode("utf-8", errors="replace")[:500] except Exception: # best-effort diagnostics body = "" suffix = f" body={body}" if body else "" return f"HTTP {error.code} {error.reason}{suffix}" if isinstance(error, urllib.error.URLError): return f"URL error: {error.reason}" return f"{error.__class__.__name__}: {error}"
def with_retry(operation, attempts: int = 5, label: str = "request", refresh_on_expired_token: Callable[[], None] | None = None): for attempt in range(1, attempts + 1): try: return operation() except urllib.error.HTTPError as error: summary = error_summary(error) expired_token = error.code == 400 and "ExpiredToken" in summary retryable = error.code == 429 or 500 <= error.code <= 599 or expired_token print(f"[{label}] attempt {attempt}/{attempts} failed: {summary}", file=sys.stderr, flush=True) if not retryable or attempt == attempts: raise if expired_token and refresh_on_expired_token: refresh_on_expired_token() except urllib.error.URLError as error: print(f"[{label}] attempt {attempt}/{attempts} failed: {error_summary(error)}", file=sys.stderr, flush=True) if attempt == attempts: raise except (TimeoutError, socket.timeout) as error: print(f"[{label}] attempt {attempt}/{attempts} failed: {error_summary(error)}", file=sys.stderr, flush=True) if attempt == attempts: raise time.sleep(min(2 ** (attempt - 1), 16)) raise RuntimeError("retry loop exhausted")
def get_json(url: str, timeout_env: str = "ATP_REQUEST_TIMEOUT") -> dict: def request(): with urllib.request.urlopen(url, timeout=request_timeout(timeout_env)) as resp: # nosec - intentional CLI fetch return json.loads(resp.read().decode("utf-8"))
return with_retry(request, label=f"GET {url}")
def post_json( url: str, payload: dict, token: str | None = None, timeout_env: str = "ATP_REQUEST_TIMEOUT", token_provider: Callable[[], str | None] | None = None, refresh_on_expired_token: Callable[[], None] | None = None,) -> dict: body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
def request(): request_token = token_provider() if token_provider else token headers = {"Content-Type": "application/json"} if request_token: headers["Authorization"] = f"Bearer {request_token}" req = urllib.request.Request(url, data=body, headers=headers, method="POST") with urllib.request.urlopen(req, timeout=request_timeout(timeout_env)) as resp: # nosec - intentional CLI publish return json.loads(resp.read().decode("utf-8"))
return with_retry(request, label=f"POST {url}", refresh_on_expired_token=refresh_on_expired_token)
def post_bytes(url: str, body: bytes, token: str, mime_type: str, timeout_env: str = "ATP_REQUEST_TIMEOUT") -> dict: req = urllib.request.Request(url, data=body, headers={"Authorization": f"Bearer {token}", "Content-Type": mime_type}, method="POST")
def request(): with urllib.request.urlopen(req, timeout=request_timeout(timeout_env)) as resp: # nosec - intentional CLI publish return json.loads(resp.read().decode("utf-8"))
return with_retry(request, label=f"POST {url}")
def get_bytes(url: str, timeout_env: str = "ATP_REQUEST_TIMEOUT") -> tuple[bytes, str | None]: def request(): with urllib.request.urlopen(url, timeout=request_timeout(timeout_env)) as resp: # nosec - intentional CLI fetch content_type = resp.headers.get_content_type() return resp.read(), content_type
return with_retry(request, label=f"GET {url}")
def resolve_pds(did: str, timeout_env: str = "ATP_REQUEST_TIMEOUT") -> str: doc = get_json(f"https://plc.directory/{urllib.parse.quote(did, safe='')}", timeout_env=timeout_env) for service in doc.get("service", []): if service.get("type") == "AtprotoPersonalDataServer": return str(service["serviceEndpoint"]).rstrip("/") raise RuntimeError(f"Could not resolve PDS for {did}")
def auth_token(auth_pds: str, identifier: str, app_password: str, timeout_env: str = "ATP_REQUEST_TIMEOUT") -> str: response = post_json( f"{auth_pds}/xrpc/com.atproto.server.createSession", {"identifier": identifier, "password": app_password}, timeout_env=timeout_env, ) token = response.get("accessJwt") if not token: raise RuntimeError("createSession response did not include accessJwt") return str(token)
def refresh_auth_token(repo: str, pds: str, timeout_env: str = "ATP_REQUEST_TIMEOUT") -> str | None: identifier = os.environ.get("ATP_IDENTIFIER", repo) app_password = os.environ.get("ATP_APP_PASSWORD") or os.environ.get("ATP_PASSWORD") if not app_password: print("ATP_APP_PASSWORD is required to refresh expired ATP_AUTH_TOKEN", file=sys.stderr) return None auth_pds = os.environ.get("ATP_AUTH_PDS_URL", "").rstrip("/") or pds print(f"Refreshing expired token for {identifier} via {auth_pds}", file=sys.stderr, flush=True) return auth_token(auth_pds, identifier, app_password, timeout_env=timeout_env)
def get_or_create_auth_token(repo: str, pds: str, timeout_env: str = "ATP_REQUEST_TIMEOUT") -> str | None: token = os.environ.get("ATP_AUTH_TOKEN") if token: return token identifier = os.environ.get("ATP_IDENTIFIER", repo) app_password = os.environ.get("ATP_APP_PASSWORD") or os.environ.get("ATP_PASSWORD") if not app_password: print("ATP_APP_PASSWORD is required when ATP_AUTH_TOKEN is not set", file=sys.stderr) return None auth_pds = os.environ.get("ATP_AUTH_PDS_URL", "").rstrip("/") or pds print(f"Authenticating {identifier} via {auth_pds}") return auth_token(auth_pds, identifier, app_password, timeout_env=timeout_env)
def list_records(pds: str, repo: str, collection: str, limit: int = 100, timeout_env: str = "ATP_REQUEST_TIMEOUT") -> list[dict]: records = [] cursor = None while True: params = {"repo": repo, "collection": collection, "limit": str(limit)} if cursor: params["cursor"] = cursor url = f"{pds}/xrpc/com.atproto.repo.listRecords?{urllib.parse.urlencode(params)}" page = get_json(url, timeout_env=timeout_env) records.extend(page.get("records", [])) cursor = page.get("cursor") if not cursor: return records
def compact(obj): if isinstance(obj, dict): return {k: compact(v) for k, v in obj.items() if v is not None} if isinstance(obj, list): return [compact(v) for v in obj] return obj
def put_record(pds: str, token: str, repo: str, collection: str, key: str, record: dict, timeout_env: str = "ATP_REQUEST_TIMEOUT") -> dict: current_token = token
def token_provider() -> str: return current_token
def refresh_token() -> None: nonlocal current_token refreshed = refresh_auth_token(repo, pds, timeout_env=timeout_env) if refreshed: current_token = refreshed
return post_json( f"{pds}/xrpc/com.atproto.repo.putRecord", {"repo": repo, "collection": collection, "rkey": key, "record": compact(record)}, token_provider=token_provider, refresh_on_expired_token=refresh_token, timeout_env=timeout_env, )
def rkey_from_uri(uri: str) -> str: return uri.rstrip("/").split("/")[-1]