From 9eadee452c66767346eb8cbb6839fc217afa141f Mon Sep 17 00:00:00 2001 From: Aly Raffauf Date: Mon, 06 Apr 2026 14:28:09 +0000 Subject: [PATCH] requests -> httpx --- identity.py | 186 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------------------------------------------------------------------------------------------ main.py | 48 +++++++++++++++++++++++++----------------------- pyproject.toml | 1 + uv.lock | 2 ++ 4 file(s) changed, 124 insertion(s)(+), 113 deletion(s)(-) diff --git a/identity.py b/identity.py --- a/identity.py +++ b/identity.py @@ -1,7 +1,7 @@ +import asyncio import time -from concurrent.futures import ThreadPoolExecutor -import requests +import httpx CONSTELLATION_URL = "https://constellation.microcosm.blue" SLINGSHOT_URL = "https://slingshot.microcosm.blue" @@ -16,24 +16,25 @@ _recent_bites_ttl = 60 -def resolve_did(identifier: str) -> str | None: - """Resolve a handle to a DID via Slingshot. Returns the DID, or None if resolution fails.""" +async def resolve_did(identifier: str) -> str | None: + """Resolve a handle to a DID via Slingshot.""" if identifier.startswith("did:"): return identifier try: - resp = requests.get( - f"{SLINGSHOT_URL}/xrpc/com.atproto.identity.resolveHandle", - params={"handle": identifier}, - timeout=5, - ) + async with httpx.AsyncClient() as client: + resp = await client.get( + f"{SLINGSHOT_URL}/xrpc/com.atproto.identity.resolveHandle", + params={"handle": identifier}, + timeout=5, + ) if resp.status_code != 200: return None return resp.json().get("did") - except (requests.RequestException, ValueError): + except (httpx.HTTPError, ValueError): return None -def resolve_identity(did: str) -> tuple[str | None, str | None]: +async def resolve_identity(did: str) -> tuple[str | None, str | None]: """Resolve a DID to its handle and PDS URL via Slingshot.""" now = time.time() if did in _identity_cache: @@ -42,16 +43,17 @@ return result try: - resp = requests.get( - f"{SLINGSHOT_URL}/xrpc/blue.microcosm.identity.resolveMiniDoc", - params={"identifier": did}, - timeout=5, - ) + async with httpx.AsyncClient() as client: + resp = await client.get( + f"{SLINGSHOT_URL}/xrpc/blue.microcosm.identity.resolveMiniDoc", + params={"identifier": did}, + timeout=5, + ) if resp.status_code != 200: _identity_cache[did] = ((None, None), now) return None, None data = resp.json() - except (requests.RequestException, ValueError): + except (httpx.HTTPError, ValueError): return None, None handle = data.get("handle") @@ -61,7 +63,7 @@ return handle, pds_url -def fetch_profile(did: str, pds_url: str) -> dict[str, str | None]: +async def fetch_profile(did: str, pds_url: str) -> dict[str, str | None]: """Fetch a user's Bluesky profile via Slingshot.""" now = time.time() if did in _profile_cache: @@ -70,19 +72,20 @@ return result try: - resp = requests.get( - f"{SLINGSHOT_URL}/xrpc/com.atproto.repo.getRecord", - params={ - "repo": did, - "collection": "app.bsky.actor.profile", - "rkey": "self", - }, - timeout=5, - ) + async with httpx.AsyncClient() as client: + resp = await client.get( + f"{SLINGSHOT_URL}/xrpc/com.atproto.repo.getRecord", + params={ + "repo": did, + "collection": "app.bsky.actor.profile", + "rkey": "self", + }, + timeout=5, + ) if resp.status_code != 200: return {} value = resp.json().get("value", {}) - except requests.RequestException, ValueError: + except (httpx.HTTPError, ValueError): return {} avatar_url = None @@ -107,7 +110,7 @@ return profile -def fetch_recent_bites(limit: int = 5) -> list[dict[str, str | None]]: +async def fetch_recent_bites(limit: int = 5) -> list[dict[str, str | None]]: """Fetch the most recent bites network-wide from UFOs.""" global _recent_bites_cache now = time.time() @@ -117,32 +120,34 @@ return cached[:limit] try: - resp = requests.get( - f"{UFOS_API_URL}/records", - params={"collection": "blue.morsels.bite", "limit": limit}, - timeout=5, - ) + async with httpx.AsyncClient() as client: + resp = await client.get( + f"{UFOS_API_URL}/records", + params={"collection": "blue.morsels.bite", "limit": limit}, + timeout=5, + ) if resp.status_code != 200: return [] raw = resp.json()[:limit] - except requests.RequestException, ValueError: + except (httpx.HTTPError, ValueError): return [] - # Resolve all identities in parallel + # Resolve all identities concurrently dids = [item.get("did", "") for item in raw] unique_dids = list(set(d for d in dids if d)) - with ThreadPoolExecutor(max_workers=5) as pool: - identity_results = dict(zip(unique_dids, pool.map(resolve_identity, unique_dids))) + identity_tasks = {did: resolve_identity(did) for did in unique_dids} + identity_values = await asyncio.gather(*identity_tasks.values()) + identity_results = dict(zip(identity_tasks.keys(), identity_values)) - # Fetch profiles in parallel too (for avatar URLs) - def _fetch_profile_for_did(did: str) -> dict[str, str | None]: + # Fetch profiles concurrently + async def _fetch_profile_for_did(did: str) -> tuple[str, dict[str, str | None]]: handle, pds_url = identity_results.get(did, (None, None)) if pds_url: - return fetch_profile(did, pds_url) - return {} + return did, await fetch_profile(did, pds_url) + return did, {} - with ThreadPoolExecutor(max_workers=5) as pool: - profile_results = dict(zip(unique_dids, pool.map(_fetch_profile_for_did, unique_dids))) + profile_values = await asyncio.gather(*[_fetch_profile_for_did(d) for d in unique_dids]) + profile_results = dict(profile_values) bites = [] for item in raw: @@ -167,62 +172,63 @@ return bites -def fetch_replies(did: str, rkey: str) -> list[dict[str, str]]: +async def fetch_replies(did: str, rkey: str) -> list[dict[str, str]]: """Fetch reply backlinks from Constellation.""" at_uri = f"at://{did}/blue.morsels.bite/{rkey}" try: - resp = requests.get( - f"{CONSTELLATION_URL}/xrpc/blue.microcosm.links.getBacklinks", - params={ - "subject": at_uri, - "source": "blue.morsels.reply:subject.uri", - "limit": 100, - }, - timeout=5, - ) - if resp.status_code != 200: - return [] - return resp.json().get("records", []) - except requests.RequestException, ValueError: - return [] - - -def hydrate_replies(records: list[dict[str, str]]) -> list[dict[str, str | None]]: - """Fetch reply record contents from Slingshot.""" - replies: list[dict[str, str | None]] = [] - for record in records: - did = record.get("did") - rkey = record.get("rkey") - if not did or not rkey: - continue - - try: - resp = requests.get( - f"{SLINGSHOT_URL}/xrpc/com.atproto.repo.getRecord", + async with httpx.AsyncClient() as client: + resp = await client.get( + f"{CONSTELLATION_URL}/xrpc/blue.microcosm.links.getBacklinks", params={ - "repo": did, - "collection": "blue.morsels.reply", - "rkey": rkey, + "subject": at_uri, + "source": "blue.morsels.reply:subject.uri", + "limit": 100, }, timeout=5, ) + if resp.status_code != 200: + return [] + return resp.json().get("records", []) + except (httpx.HTTPError, ValueError): + return [] + + +async def hydrate_replies(records: list[dict[str, str]]) -> list[dict[str, str | None]]: + """Fetch reply record contents from Slingshot concurrently.""" + + async def _hydrate_one(record: dict[str, str]) -> dict[str, str | None] | None: + did = record.get("did") + rkey = record.get("rkey") + if not did or not rkey: + return None + + try: + async with httpx.AsyncClient() as client: + resp = await client.get( + f"{SLINGSHOT_URL}/xrpc/com.atproto.repo.getRecord", + params={ + "repo": did, + "collection": "blue.morsels.reply", + "rkey": rkey, + }, + timeout=5, + ) if resp.status_code != 200: - continue + return None value = resp.json().get("value", {}) - except requests.RequestException, ValueError: - continue + except (httpx.HTTPError, ValueError): + return None - handle, _ = resolve_identity(did) + handle, _ = await resolve_identity(did) - replies.append( - { - "did": did, - "handle": handle, - "rkey": rkey, - "text": value.get("text", ""), - "created_at": value.get("createdAt", ""), - } - ) + return { + "did": did, + "handle": handle, + "rkey": rkey, + "text": value.get("text", ""), + "created_at": value.get("createdAt", ""), + } - return replies + results = await asyncio.gather(*[_hydrate_one(r) for r in records]) + return [r for r in results if r is not None] diff --git a/main.py b/main.py --- a/main.py +++ b/main.py @@ -11,6 +11,7 @@ from typing import Any from urllib.parse import urlencode, urlparse +import httpx import regex import requests from atproto import Client, models @@ -126,26 +127,27 @@ return Response(data, mimetype=content_type) # Resolve PDS to find avatar - handle, pds_url = resolve_identity(did) + handle, pds_url = await resolve_identity(did) if pds_url is None: return Response(status=404) - profile = fetch_profile(did, pds_url) + profile = await fetch_profile(did, pds_url) cdn_url = profile.get("avatar_url") blob_url = profile.get("avatar_blob_url") if not cdn_url and not blob_url: return Response(status=404) resp = None - for url in [cdn_url, blob_url]: - if not url: - continue - try: - resp = requests.get(url, timeout=5) - if resp.status_code == 200: - break - except (requests.RequestException, ValueError): - continue + async with httpx.AsyncClient() as http: + for url in [cdn_url, blob_url]: + if not url: + continue + try: + resp = await http.get(url, timeout=5) + if resp.status_code == 200: + break + except (httpx.HTTPError, ValueError): + continue if resp is None or resp.status_code != 200: return Response(status=502) @@ -176,7 +178,7 @@ return highlight(content or "", lexer, formatter).rstrip("\n") -def require_identity( +async def require_identity( identifier: str, redirect_endpoint: str, **redirect_kwargs: Any ) -> tuple[str, str | None, str, dict] | WerkzeugResponse: """Resolve an identifier to a DID, redirecting handles to canonical DID URLs. @@ -184,19 +186,19 @@ Returns (did, handle, pds_url, profile) or redirects/aborts. Callers must check the return — if it's a Response (redirect), return it directly. """ - did = resolve_did(identifier) + did = await resolve_did(identifier) if did is None: abort(404, "User not found") if did != identifier: return redirect(url_for(redirect_endpoint, identifier=did, **redirect_kwargs)) - handle, pds_url = resolve_identity(did) + handle, pds_url = await resolve_identity(did) if handle is None and pds_url is None: abort(404, "User not found.") if pds_url is None: abort(502, "Could not reach this user's server.") - profile = fetch_profile(did, pds_url) + profile = await fetch_profile(did, pds_url) return did, handle, pds_url, profile @@ -446,12 +448,12 @@ if username.startswith("@"): username = username[1:] - did = resolve_did(username) + did = await resolve_did(username) if did is None: flash("Could not find that account. Check your handle and try again.", "error") return redirect(url_for("index")) - handle, pds_url = resolve_identity(did) + handle, pds_url = await resolve_identity(did) if pds_url is None: flash("Could not reach your server. Try again later.", "error") return redirect(url_for("index")) @@ -611,7 +613,7 @@ @app.route("/") async def index() -> str: - recent = fetch_recent_bites(limit=5) + recent = await fetch_recent_bites(limit=5) # Pre-populate bite cache from feed data now = time.time() @@ -666,7 +668,7 @@ @app.route("/u/") async def list_bites(identifier: str) -> WerkzeugResponse | str: - result = require_identity(identifier, "list_bites") + result = await require_identity(identifier, "list_bites") if not isinstance(result, tuple): return result did, handle, pds_url, profile = result @@ -684,7 +686,7 @@ @app.route("/b//") async def view_bite(identifier: str, rkey: str) -> WerkzeugResponse | str: - result = require_identity(identifier, "view_bite", rkey=rkey) + result = await require_identity(identifier, "view_bite", rkey=rkey) if not isinstance(result, tuple): return result did, handle, pds_url, profile = result @@ -727,8 +729,8 @@ content = bite["content"] created_at = bite["created_at"] - raw_replies = fetch_replies(did, rkey) - replies = hydrate_replies(raw_replies) + raw_replies = await fetch_replies(did, rkey) + replies = await hydrate_replies(raw_replies) pending = session.pop("pending_reply", None) if pending: @@ -808,7 +810,7 @@ return redirect(url_for("oauth_login")) await check_csrf() - if resolve_did(identifier) != g.user["did"]: + if await resolve_did(identifier) != g.user["did"]: abort(403, "You can only delete your own bites.") resp = delete_record(COLLECTION, rkey) diff --git a/pyproject.toml b/pyproject.toml --- a/pyproject.toml +++ b/pyproject.toml @@ -8,6 +8,7 @@ "atproto>=0.0.65", "authlib>=1.6.9", "flask>=3.1.3", + "httpx>=0.28.1", "hypercorn>=0.18.0", "pygments>=2.20.0", "quart>=0.20.0", diff --git a/uv.lock b/uv.lock --- a/uv.lock +++ b/uv.lock @@ -427,6 +427,7 @@ { name = "atproto" }, { name = "authlib" }, { name = "flask" }, + { name = "httpx" }, { name = "hypercorn" }, { name = "pygments" }, { name = "quart" }, @@ -439,6 +440,7 @@ { name = "atproto", specifier = ">=0.0.65" }, { name = "authlib", specifier = ">=1.6.9" }, { name = "flask", specifier = ">=3.1.3" }, + { name = "httpx", specifier = ">=0.28.1" }, { name = "hypercorn", specifier = ">=0.18.0" }, { name = "pygments", specifier = ">=2.20.0" }, { name = "quart", specifier = ">=0.20.0" }, -- tangled.sh