feat(atproto): resolve DID to handle and display name via enrichment UDFs #418
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,167 @@ | ||
| """Optional enrichment UDFs that resolve an ATProto DID to profile fields. | ||
|
|
||
| JetStream events identify the actor only by DID, which isn't searchable the way a | ||
| handle or display name is. These UDFs resolve a DID to those fields via Bluesky's | ||
| public, unauthenticated AppView (`app.bsky.actor.getProfile`). | ||
|
|
||
| They are registered by the plugin but wired into rules only via the opt-in | ||
| `models/enrichment.sml`, because each unique DID costs an external API call -- | ||
| great for demos, but a dependency you don't want in a load test. The whole | ||
| profile is fetched once per DID and cached, and lookups fail soft (the feature is | ||
| simply absent) when the API errors or rate-limits. | ||
|
|
||
| Async UDFs run concurrently in a gevent pool, so a rule that reads both the handle | ||
| and the display name would fire `AtprotoHandle` and `AtprotoDisplayName` at the | ||
| same time. Both would miss a cold cache and each make its own `getProfile` call. | ||
| To avoid that, a fetch in progress for a DID is shared: the second greenlet waits | ||
| on the first one's result instead of making a duplicate request. Cached entries | ||
| expire after `_CACHE_TTL_SECONDS`, since handles and display names change; the | ||
| fuller approach is to bust a DID's entry when an identity or profile-update event | ||
| comes through JetStream, which is left out here to keep the example focused. | ||
|
|
||
| See the README's "Extending the enrichment" section for how to expose more of the | ||
| profile (account age, follower counts, existing labels) from the same cached fetch. | ||
| """ | ||
|
|
||
| import time | ||
| from collections import OrderedDict | ||
| from threading import Event, Lock | ||
| from typing import Any | ||
|
|
||
| import requests | ||
| from osprey.engine.executor.execution_context import ExecutionContext, ExpectedUdfException | ||
| from osprey.engine.udf.arguments import ArgumentsBase | ||
| from osprey.engine.udf.base import UDFBase | ||
|
|
||
| _ATPROTO_CATEGORY = 'ATProto' | ||
| _GET_PROFILE_URL = 'https://public.api.bsky.app/xrpc/app.bsky.actor.getProfile' | ||
| _REQUEST_TIMEOUT_SECONDS = 5 | ||
| _CACHE_MAX_SIZE = 100_000 | ||
| _CACHE_TTL_SECONDS = 60 * 60 | ||
|
|
||
| _session = requests.Session() | ||
| # did -> (profile dict, monotonic time at which the entry expires). Ordered so the | ||
| # least-recently-used entry is evicted first once the cache is full. | ||
| _profile_cache: 'OrderedDict[str, tuple[dict[str, Any], float]]' = OrderedDict() | ||
| # did -> a fetch currently in progress, so concurrent misses for the same DID | ||
| # (e.g. AtprotoHandle and AtprotoDisplayName on one event) share one API call. | ||
| _inflight: dict[str, '_InflightFetch'] = {} | ||
| _cache_lock = Lock() | ||
|
|
||
|
|
||
| class _InflightFetch: | ||
| """A single `getProfile` call in progress, shared by every greenlet awaiting it. | ||
|
|
||
| The greenlet that created it does the fetch and populates `profile` or `error` | ||
| before setting `done`; waiters block on `done`, then read the result. Under | ||
| gevent's cooperative scheduling this needs no memory barrier -- the waiter only | ||
| runs after `done.set()` yields back to it. | ||
| """ | ||
|
|
||
| __slots__ = ('done', 'profile', 'error') | ||
|
|
||
| def __init__(self) -> None: | ||
| self.done = Event() | ||
| self.profile: dict[str, Any] | None = None | ||
| self.error: Exception | None = None | ||
|
|
||
|
|
||
| def _get_profile_from_api(did: str) -> dict[str, Any]: | ||
| """Hit the public getProfile endpoint. Raises on any transport/HTTP/parse error.""" | ||
| response = _session.get(_GET_PROFILE_URL, params={'actor': did}, timeout=_REQUEST_TIMEOUT_SECONDS) | ||
| response.raise_for_status() | ||
| profile = response.json() | ||
| if not isinstance(profile, dict): | ||
| raise ValueError('getProfile did not return an object') | ||
| return profile | ||
|
|
||
|
|
||
| def _fetch_profile(did: str) -> dict[str, Any]: | ||
| """Return the getProfile response for a DID, coalescing concurrent cache misses. | ||
|
|
||
| Raises on any transport/HTTP/parse error so callers can fail soft. | ||
| """ | ||
| with _cache_lock: | ||
| cached = _profile_cache.get(did) | ||
| if cached is not None: | ||
| profile, expires_at = cached | ||
| if time.monotonic() < expires_at: | ||
| _profile_cache.move_to_end(did) | ||
| return profile | ||
| del _profile_cache[did] | ||
|
|
||
| inflight = _inflight.get(did) | ||
| is_leader = inflight is None | ||
| if inflight is None: | ||
| inflight = _InflightFetch() | ||
| _inflight[did] = inflight | ||
|
|
||
| if not is_leader: | ||
| # Someone else is already fetching this DID; wait for their result. | ||
| inflight.done.wait() | ||
| if inflight.error is not None: | ||
| raise inflight.error | ||
| assert inflight.profile is not None | ||
| return inflight.profile | ||
|
|
||
| # Leader: do the request outside the lock so a slow call doesn't block other DIDs. | ||
| try: | ||
| profile = _get_profile_from_api(did) | ||
| except Exception as exc: | ||
| inflight.error = exc | ||
| with _cache_lock: | ||
| _inflight.pop(did, None) | ||
| inflight.done.set() | ||
| raise | ||
|
|
||
| with _cache_lock: | ||
| _profile_cache[did] = (profile, time.monotonic() + _CACHE_TTL_SECONDS) | ||
| _profile_cache.move_to_end(did) | ||
| while len(_profile_cache) > _CACHE_MAX_SIZE: | ||
| _profile_cache.popitem(last=False) | ||
| _inflight.pop(did, None) | ||
| inflight.profile = profile | ||
| inflight.done.set() | ||
| return profile | ||
|
|
||
|
|
||
| def _profile_or_skip(did: str) -> dict[str, Any]: | ||
| """Fetch the cached profile, converting any lookup failure into a soft skip.""" | ||
| try: | ||
| return _fetch_profile(did) | ||
| except (requests.RequestException, ValueError): | ||
| raise ExpectedUdfException() | ||
|
|
||
|
|
||
| class DidArguments(ArgumentsBase): | ||
| did: str | ||
| """The ATProto DID to resolve (e.g. the actor's `$.did`).""" | ||
|
|
||
|
|
||
| class AtprotoHandle(UDFBase[DidArguments, str]): | ||
| """Resolves an ATProto DID to its current handle.""" | ||
|
|
||
| category = _ATPROTO_CATEGORY | ||
| execute_async = True | ||
|
|
||
| def execute(self, execution_context: ExecutionContext, arguments: DidArguments) -> str: | ||
| handle = _profile_or_skip(arguments.did).get('handle') | ||
| # getProfile is untrusted JSON, so guard the type: a truthy non-string | ||
| # would otherwise cross the str return boundary. Treat it like a missing | ||
| # field and skip. | ||
| if not isinstance(handle, str) or not handle: | ||
|
Comment on lines
+149
to
+152
Contributor
There was a problem hiding this comment. 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win 🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -eu
printf '%s\n' '--- candidate files ---'
git ls-files | grep -E '(^|/)(enrichment_udfs|test.*atproto|.*atproto.*test|tests?)/' | head -200
printf '%s\n' '--- target symbols and references ---'
rg -n --glob '*.py' 'getProfile|displayName|ExpectedUdfException|enrichment_udfs' .
printf '%s\n' '--- target structure ---'
ast-grep outline example_atproto_plugins/src/atproto_plugin/enrichment_udfs.pyRepository: roostorg/osprey Length of output: 33501 🏁 Script executed: #!/bin/bash
set -eu
printf '%s\n' '--- test file ---'
cat -n example_atproto_plugins/tests/test_enrichment_udfs.py
printf '%s\n' '--- implementation under review ---'
sed -n '120,180p' example_atproto_plugins/src/atproto_plugin/enrichment_udfs.py
printf '%s\n' '--- project test configuration ---'
rg -n --glob 'pyproject.toml' --glob 'tox.ini' --glob 'pytest.ini' --glob 'conftest.py' 'example_atproto|pytest|testpaths|pythonpath' .Repository: roostorg/osprey Length of output: 15516 🏁 Script executed: #!/bin/bash
set -eu
python3 - <<'PY'
import ast
from pathlib import Path
test_path = Path("example_atproto_plugins/tests/test_enrichment_udfs.py")
tree = ast.parse(test_path.read_text())
print("--- execute coverage ---")
for node in ast.walk(tree):
if isinstance(node, ast.FunctionDef) and node.name.startswith("test_"):
source = ast.get_source_segment(test_path.read_text(), node) or ""
if ".execute(" in source:
print(node.name)
print("--- non-string literals in test calls ---")
for node in ast.walk(tree):
if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) and node.func.attr == "execute":
values = []
for ancestor in ast.walk(node):
if isinstance(ancestor, ast.Constant) and not isinstance(ancestor.value, str):
values.append(repr(ancestor.value))
print(ast.get_source_segment(test_path.read_text(), node).splitlines()[0], "=>", values)
print("--- direct field fixtures ---")
for node in ast.walk(tree):
if isinstance(node, ast.Dict):
keys = [
key.value for key in node.keys
if isinstance(key, ast.Constant) and isinstance(key.value, str)
]
if "handle" in keys or "displayName" in keys:
print(
ast.get_source_segment(test_path.read_text(), node).replace("\n", " ")
)
PYRepository: roostorg/osprey Length of output: 864 Add regression tests for non-string profile fields. Test non-string 🤖 Prompt for AI AgentsSource: Coding guidelines |
||
| raise ExpectedUdfException() | ||
| return handle | ||
|
Comment on lines
+147
to
+154
Member
There was a problem hiding this comment. hmm. this is interesting and im not exactly sure how to best do this with SML. it's probably most likely that we're going to want both the handle and the DID in whatever model deals with user events. in that case, especially since these are going to get executed async, i think we'll end up with two requests getting made for the handle. the current locking won't actually stop a second request for being made for a user. this, i think, will double the amount of requests we are making to the api. there's a couple of things that we could probably do, but the easiest might just be to do something like:
i think this will work with SML? but im not sure... |
||
|
|
||
|
|
||
| class AtprotoDisplayName(UDFBase[DidArguments, str]): | ||
| """Resolves an ATProto DID to its display name.""" | ||
|
|
||
| category = _ATPROTO_CATEGORY | ||
| execute_async = True | ||
|
|
||
| def execute(self, execution_context: ExecutionContext, arguments: DidArguments) -> str: | ||
| display_name = _profile_or_skip(arguments.did).get('displayName') | ||
| if not isinstance(display_name, str) or not display_name: | ||
| raise ExpectedUdfException() | ||
| return display_name | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.