Refresh a token if needed

chitose/agent.py

@@ -3,6 +3,7 @@
 from datetime import timezone
 import typing
 import json
+import urllib
 
 from chitose.app.bsky.feed.like import Like
 from chitose.app.bsky.feed.post import Post
@@ -12,21 +13,58 @@
 
 from .app import App_
 from .com import Com_
+from .xrpc import call
+from .xrpc import XrpcParams
+from .xrpc import XrpcData
+from .xrpc import XrpcHeaders
 
 
 class BskyAgent:
     def __init__(self, service: str) -> None:
         self.service = service
-        self.headers: dict[str, str] = {}
-        self.session: dict = {}
+        self.session: dict[str, str] = {}
 
-    @property
+    def _add_auth_header(self, headers: dict[str, str]):
+        if 'authorization' not in headers and 'accessJwt' in self.session:
+            return headers | {
+                'authorization': f'Bearer {self.session["accessJwt"]}'
+            }
+
+        return headers
+
+    def _refresh_session(self, e: urllib.error.HTTPError):
+        if 'refreshJwt' not in self.session:
+            raise e
+
+        error = json.loads(e.read().decode())['error']
+        if error != 'ExpiredToken':
+            raise e
+
+        self.session = json.loads(
+            call('com.atproto.server.refreshSession',
+                 [], {}, self.service,
+                 {'authorization':
+                  f'Bearer {self.session["refreshJwt"]}'})
+        )
+
+    def _call(self, method: str, params: XrpcParams,
+              d: XrpcData, headers: XrpcHeaders) -> bytes:
+        try:
+            return call(method, params, d, self.service,
+                        self._add_auth_header(headers))
+        except urllib.error.HTTPError as e:
+            self._refresh_session(e)
+
+            return call(method, params, d, self.service,
+                        self._add_auth_header(headers))
+
+    @ property
     def app(self):
-        return App_(self.service, self.headers)
+        return App_(self._call)
 
-    @property
+    @ property
     def com(self):
-        return Com_(self.service, self.headers)
+        return Com_(self._call)
 
     def get_timeline(self, **kwargs: dict[str, typing.Any]) -> bytes:
         """See :doc:`chitose.app.bsky.feed` for available arguments."""
@@ -199,7 +237,3 @@ def login(self, identifier: str, password: str) -> None:
         self.session = json.loads(
             self.com.atproto.server.create_session(
                 identifier=identifier, password=password))
-
-        if 'accessJwt' in self.session:
-            self.headers['authorization'] \
-                = f'Bearer {self.session["accessJwt"]}'

chitose/app/__init__.py

@@ -1,14 +1,14 @@
 # GENERATED CODE - DO NOT MODIFY
 from __future__ import annotations
+from chitose.xrpc import XrpcCallable
 from .bsky import Bsky_
 
 class App_:
     """We recommend calling methods in this class via the :doc:`chitose.BskyAgent <chitose>` class instead of creating instances of this class directly."""
 
-    def __init__(self, service: str, headers: dict[str, str]) -> None:
-        self.service = service
-        self.headers = headers
+    def __init__(self, call: XrpcCallable) -> None:
+        self.call = call
 
     @property
     def bsky(self):
-        return Bsky_(self.service, self.headers)
+        return Bsky_(self.call)

chitose/app/bsky/__init__.py

@@ -1,5 +1,6 @@
 # GENERATED CODE - DO NOT MODIFY
 from __future__ import annotations
+from chitose.xrpc import XrpcCallable
 from .actor import Actor_
 from .embed import Embed_
 from .feed import Feed_
@@ -11,34 +12,33 @@
 class Bsky_:
     """We recommend calling methods in this class via the :doc:`chitose.BskyAgent <chitose>` class instead of creating instances of this class directly."""
 
-    def __init__(self, service: str, headers: dict[str, str]) -> None:
-        self.service = service
-        self.headers = headers
+    def __init__(self, call: XrpcCallable) -> None:
+        self.call = call
 
     @property
     def actor(self):
-        return Actor_(self.service, self.headers)
+        return Actor_(self.call)
 
     @property
     def embed(self):
-        return Embed_(self.service, self.headers)
+        return Embed_(self.call)
 
     @property
     def feed(self):
-        return Feed_(self.service, self.headers)
+        return Feed_(self.call)
 
     @property
     def graph(self):
-        return Graph_(self.service, self.headers)
+        return Graph_(self.call)
 
     @property
     def notification(self):
-        return Notification_(self.service, self.headers)
+        return Notification_(self.call)
 
     @property
     def richtext(self):
-        return Richtext_(self.service, self.headers)
+        return Richtext_(self.call)
 
     @property
     def unspecced(self):
-        return Unspecced_(self.service, self.headers)
+        return Unspecced_(self.call)

chitose/app/bsky/actor/__init__.py

@@ -1,5 +1,6 @@
 # GENERATED CODE - DO NOT MODIFY
 from __future__ import annotations
+from chitose.xrpc import XrpcCallable
 from .get_preferences import _get_preferences
 from .get_profile import _get_profile
 from .get_profiles import _get_profiles
@@ -13,34 +14,33 @@
 class Actor_:
     """We recommend calling methods in this class via the :doc:`chitose.BskyAgent <chitose>` class instead of creating instances of this class directly."""
 
-    def __init__(self, service: str, headers: dict[str, str]) -> None:
-        self.service = service
-        self.headers = headers
+    def __init__(self, call: XrpcCallable) -> None:
+        self.call = call
 
     def search_actors_typeahead(self, term: typing.Optional[str]=None, limit: typing.Optional[int]=None) -> bytes:
         """Find actor suggestions for a search term."""
-        return _search_actors_typeahead(self.service, self.headers, term, limit)
+        return _search_actors_typeahead(self.call, term, limit)
 
     def put_preferences(self, preferences: chitose.app.bsky.actor.defs.Preferences) -> bytes:
         """Sets the private preferences attached to the account."""
-        return _put_preferences(self.service, self.headers, preferences)
+        return _put_preferences(self.call, preferences)
 
     def get_profile(self, actor: str) -> bytes:
         """"""
-        return _get_profile(self.service, self.headers, actor)
+        return _get_profile(self.call, actor)
 
     def get_suggestions(self, limit: typing.Optional[int]=None, cursor: typing.Optional[str]=None) -> bytes:
         """Get a list of actors suggested for following. Used in discovery UIs."""
-        return _get_suggestions(self.service, self.headers, limit, cursor)
+        return _get_suggestions(self.call, limit, cursor)
 
     def search_actors(self, term: typing.Optional[str]=None, limit: typing.Optional[int]=None, cursor: typing.Optional[str]=None) -> bytes:
         """Find actors matching search criteria."""
-        return _search_actors(self.service, self.headers, term, limit, cursor)
+        return _search_actors(self.call, term, limit, cursor)
 
     def get_profiles(self, actors: list[str]) -> bytes:
         """"""
-        return _get_profiles(self.service, self.headers, actors)
+        return _get_profiles(self.call, actors)
 
     def get_preferences(self) -> bytes:
         """Get private preferences attached to the account."""
-        return _get_preferences(self.service, self.headers)
+        return _get_preferences(self.call)

chitose/app/bsky/actor/get_preferences.py

@@ -3,6 +3,6 @@
 from __future__ import annotations
 import chitose
 
-def _get_preferences(service: str, headers: dict[str, str]) -> bytes:
+def _get_preferences(call: chitose.xrpc.XrpcCallable) -> bytes:
     """Get private preferences attached to the account."""
-    return chitose.xrpc.call('app.bsky.actor.getPreferences', [], None, service, {} | headers)
+    return call('app.bsky.actor.getPreferences', [], None, {})

chitose/app/bsky/actor/get_profile.py

@@ -3,6 +3,6 @@
 from __future__ import annotations
 import chitose
 
-def _get_profile(service: str, headers: dict[str, str], actor: str) -> bytes:
+def _get_profile(call: chitose.xrpc.XrpcCallable, actor: str) -> bytes:
     """"""
-    return chitose.xrpc.call('app.bsky.actor.getProfile', [('actor', actor)], None, service, {} | headers)
+    return call('app.bsky.actor.getProfile', [('actor', actor)], None, {})

chitose/app/bsky/actor/get_profiles.py

@@ -3,6 +3,6 @@
 from __future__ import annotations
 import chitose
 
-def _get_profiles(service: str, headers: dict[str, str], actors: list[str]) -> bytes:
+def _get_profiles(call: chitose.xrpc.XrpcCallable, actors: list[str]) -> bytes:
     """"""
-    return chitose.xrpc.call('app.bsky.actor.getProfiles', [('actors', actors)], None, service, {} | headers)
+    return call('app.bsky.actor.getProfiles', [('actors', actors)], None, {})

chitose/app/bsky/actor/get_suggestions.py

@@ -4,6 +4,6 @@
 import chitose
 import typing
 
-def _get_suggestions(service: str, headers: dict[str, str], limit: typing.Optional[int]=None, cursor: typing.Optional[str]=None) -> bytes:
+def _get_suggestions(call: chitose.xrpc.XrpcCallable, limit: typing.Optional[int]=None, cursor: typing.Optional[str]=None) -> bytes:
     """Get a list of actors suggested for following. Used in discovery UIs."""
-    return chitose.xrpc.call('app.bsky.actor.getSuggestions', [('limit', limit), ('cursor', cursor)], None, service, {} | headers)
+    return call('app.bsky.actor.getSuggestions', [('limit', limit), ('cursor', cursor)], None, {})

chitose/app/bsky/actor/put_preferences.py

@@ -4,6 +4,6 @@
 import chitose
 import chitose.app.bsky.actor.defs
 
-def _put_preferences(service: str, headers: dict[str, str], preferences: chitose.app.bsky.actor.defs.Preferences) -> bytes:
+def _put_preferences(call: chitose.xrpc.XrpcCallable, preferences: chitose.app.bsky.actor.defs.Preferences) -> bytes:
     """Sets the private preferences attached to the account."""
-    return chitose.xrpc.call('app.bsky.actor.putPreferences', [], {'preferences': preferences}, service, {'Content-Type': 'application/json'} | headers)
+    return call('app.bsky.actor.putPreferences', [], {'preferences': preferences}, {'Content-Type': 'application/json'})

chitose/app/bsky/actor/search_actors.py

@@ -4,6 +4,6 @@
 import chitose
 import typing
 
-def _search_actors(service: str, headers: dict[str, str], term: typing.Optional[str]=None, limit: typing.Optional[int]=None, cursor: typing.Optional[str]=None) -> bytes:
+def _search_actors(call: chitose.xrpc.XrpcCallable, term: typing.Optional[str]=None, limit: typing.Optional[int]=None, cursor: typing.Optional[str]=None) -> bytes:
     """Find actors matching search criteria."""
-    return chitose.xrpc.call('app.bsky.actor.searchActors', [('term', term), ('limit', limit), ('cursor', cursor)], None, service, {} | headers)
+    return call('app.bsky.actor.searchActors', [('term', term), ('limit', limit), ('cursor', cursor)], None, {})

chitose/app/bsky/actor/search_actors_typeahead.py

@@ -4,6 +4,6 @@
 import chitose
 import typing
 
-def _search_actors_typeahead(service: str, headers: dict[str, str], term: typing.Optional[str]=None, limit: typing.Optional[int]=None) -> bytes:
+def _search_actors_typeahead(call: chitose.xrpc.XrpcCallable, term: typing.Optional[str]=None, limit: typing.Optional[int]=None) -> bytes:
     """Find actor suggestions for a search term."""
-    return chitose.xrpc.call('app.bsky.actor.searchActorsTypeahead', [('term', term), ('limit', limit)], None, service, {} | headers)
+    return call('app.bsky.actor.searchActorsTypeahead', [('term', term), ('limit', limit)], None, {})

chitose/app/bsky/embed/__init__.py

@@ -1,9 +1,9 @@
 # GENERATED CODE - DO NOT MODIFY
 from __future__ import annotations
+from chitose.xrpc import XrpcCallable
 
 class Embed_:
     """We recommend calling methods in this class via the :doc:`chitose.BskyAgent <chitose>` class instead of creating instances of this class directly."""
 
-    def __init__(self, service: str, headers: dict[str, str]) -> None:
-        self.service = service
-        self.headers = headers
+    def __init__(self, call: XrpcCallable) -> None:
+        self.call = call

chitose/app/bsky/feed/__init__.py

@@ -1,5 +1,6 @@
 # GENERATED CODE - DO NOT MODIFY
 from __future__ import annotations
+from chitose.xrpc import XrpcCallable
 from .describe_feed_generator import _describe_feed_generator
 from .get_actor_feeds import _get_actor_feeds
 from .get_author_feed import _get_author_feed
@@ -18,54 +19,53 @@
 class Feed_:
     """We recommend calling methods in this class via the :doc:`chitose.BskyAgent <chitose>` class instead of creating instances of this class directly."""
 
-    def __init__(self, service: str, headers: dict[str, str]) -> None:
-        self.service = service
-        self.headers = headers
+    def __init__(self, call: XrpcCallable) -> None:
+        self.call = call
 
     def get_feed_generators(self, feeds: list[str]) -> bytes:
         """Get information about a list of feed generators"""
-        return _get_feed_generators(self.service, self.headers, feeds)
+        return _get_feed_generators(self.call, feeds)
 
     def get_timeline(self, algorithm: typing.Optional[str]=None, limit: typing.Optional[int]=None, cursor: typing.Optional[str]=None) -> bytes:
         """A view of the user's home timeline."""
-        return _get_timeline(self.service, self.headers, algorithm, limit, cursor)
+        return _get_timeline(self.call, algorithm, limit, cursor)
 
     def get_feed_generator(self, feed: str) -> bytes:
         """Get information about a specific feed offered by a feed generator, such as its online status"""
-        return _get_feed_generator(self.service, self.headers, feed)
+        return _get_feed_generator(self.call, feed)
 
     def get_author_feed(self, actor: str, limit: typing.Optional[int]=None, cursor: typing.Optional[str]=None) -> bytes:
         """A view of an actor's feed."""
-        return _get_author_feed(self.service, self.headers, actor, limit, cursor)
+        return _get_author_feed(self.call, actor, limit, cursor)
 
     def get_likes(self, uri: str, cid: typing.Optional[str]=None, limit: typing.Optional[int]=None, cursor: typing.Optional[str]=None) -> bytes:
         """"""
-        return _get_likes(self.service, self.headers, uri, cid, limit, cursor)
+        return _get_likes(self.call, uri, cid, limit, cursor)
 
     def get_post_thread(self, uri: str, depth: typing.Optional[int]=None) -> bytes:
         """"""
-        return _get_post_thread(self.service, self.headers, uri, depth)
+        return _get_post_thread(self.call, uri, depth)
 
     def get_reposted_by(self, uri: str, cid: typing.Optional[str]=None, limit: typing.Optional[int]=None, cursor: typing.Optional[str]=None) -> bytes:
         """"""
-        return _get_reposted_by(self.service, self.headers, uri, cid, limit, cursor)
+        return _get_reposted_by(self.call, uri, cid, limit, cursor)
 
     def describe_feed_generator(self) -> bytes:
         """Returns information about a given feed generator including TOS & offered feed URIs"""
-        return _describe_feed_generator(self.service, self.headers)
+        return _describe_feed_generator(self.call)
 
     def get_posts(self, uris: list[str]) -> bytes:
         """A view of an actor's feed."""
-        return _get_posts(self.service, self.headers, uris)
+        return _get_posts(self.call, uris)
 
     def get_feed(self, feed: str, limit: typing.Optional[int]=None, cursor: typing.Optional[str]=None) -> bytes:
         """Compose and hydrate a feed from a user's selected feed generator"""
-        return _get_feed(self.service, self.headers, feed, limit, cursor)
+        return _get_feed(self.call, feed, limit, cursor)
 
     def get_feed_skeleton(self, feed: str, limit: typing.Optional[int]=None, cursor: typing.Optional[str]=None) -> bytes:
         """A skeleton of a feed provided by a feed generator"""
-        return _get_feed_skeleton(self.service, self.headers, feed, limit, cursor)
+        return _get_feed_skeleton(self.call, feed, limit, cursor)
 
     def get_actor_feeds(self, actor: str, limit: typing.Optional[int]=None, cursor: typing.Optional[str]=None) -> bytes:
         """Retrieve a list of feeds created by a given actor"""
-        return _get_actor_feeds(self.service, self.headers, actor, limit, cursor)
+        return _get_actor_feeds(self.call, actor, limit, cursor)

chitose/app/bsky/feed/describe_feed_generator.py

@@ -4,9 +4,9 @@
 import chitose
 import typing
 
-def _describe_feed_generator(service: str, headers: dict[str, str]) -> bytes:
+def _describe_feed_generator(call: chitose.xrpc.XrpcCallable) -> bytes:
     """Returns information about a given feed generator including TOS & offered feed URIs"""
-    return chitose.xrpc.call('app.bsky.feed.describeFeedGenerator', [], None, service, {} | headers)
+    return call('app.bsky.feed.describeFeedGenerator', [], None, {})
 
 class Feed(chitose.Object):
     """"""