Lexi Winter: 1 api.timeline: factor and clean up 1 files changed, 115 insertions(+), 92 deletions(-)
Copy & paste the following snippet into your terminal to import this patchset into git:
curl -s https://lists.sr.ht/~ihabunek/toot-discuss/patches/48460/mbox | git am -3Learn more about email & git
In preparation for new timeline features, clean up api.timeline and
makes things more consistent.
The new function fetch_timeline() is now always used to do the actual
HTTP fetch. This implements the 'next_path' functionality.
Add a new StatusTimeline while is the base class for timelines that
return statuses (HomeTimeline, etc).
Clean up the 'limit' parameter; it's now always 'int | None = None'
except in fetch_timeline(), which sets the actual limit if not
specified.
Cleanup/fix a couple of Timelines to make sure they all use
fetch_timeline().
Comment out a couple of *_timline_generator() functions which weren't
used anywhere.
No functional changes intended. The external API (e.g. as used by
TimelineTab) should be unchanged.
---
Rebased on origin/main eb3c0ae5 ("ComposeScreen: support Hometown's
unfederated posts (#13)")
tooi/api/timeline.py | 207 ++++++++++++++++++++++++-------------------
1 file changed, 115 insertions(+), 92 deletions(-)
diff --git a/tooi/api/timeline.py b/tooi/api/timeline.py
index 5ecbfca..d93e46b 100644
--- a/tooi/api/timeline.py
+++ b/tooi/api/timeline.py
@@ -17,26 +17,44 @@ from tooi.data.events import Event, StatusEvent, MentionEvent, NewFollowerEvent,
from tooi.data.events import FavouriteEvent
from tooi.data.instance import InstanceInfo
from tooi.entities import Status, Notification, from_dict
-from tooi.utils.string import str_bool
Params = Optional[QueryParamTypes]
EventGenerator = AsyncGenerator[List[Event], None]
-# def anon_public_timeline_generator(instance, local: bool = False, limit: int = 40):
-# path = "/api/v1/timelines/public"
-# params = {"local": str_bool(local), "limit": limit}
-# return _anon_timeline_generator(instance, path, params)
+# Max 80, as of Mastodon 4.1.0
+DEFAULT_LIMIT = 40
-# def anon_tag_timeline_generator(instance, hashtag, local: bool = False, limit: int = 40):
-# path = f"/api/v1/timelines/tag/{quote(hashtag)}"
-# params = {"local": str_bool(local), "limit": limit}
-# return _anon_timeline_generator(instance, path, params)
+
+def _get_next_path(headers: Headers) -> str | None:
+ """Given timeline response headers, returns the path to the next batch"""
+ links = headers.get("Link", "")
+ matches = re.match(r'<([^>]+)>; rel="next"', links)
+ if matches:
+ parsed = urlparse(matches.group(1))
+ return "?".join([parsed.path, parsed.query])
+ return None
+
+
+async def fetch_timeline(
+ instance: InstanceInfo,
+ path: str,
+ params: Params | None = None,
+ limit: int | None = None):
+
+ _params = dict(params or {})
+ _params["limit"] = limit or DEFAULT_LIMIT
+
+ next_path = path
+ while next_path:
+ response = await request("GET", next_path, params=_params)
+ yield response.json()
+ next_path = _get_next_path(response.headers)
class Timeline(ABC):
"""
- Base class for a timeline.
+ Base class for a timeline. This provides some useful generators that subclasses can use.
"""
def __init__(self, name, instance: InstanceInfo):
@@ -44,21 +62,29 @@ class Timeline(ABC):
self.name = name
@abstractmethod
- def create_generator(self, limit: int = 40) -> EventGenerator:
+ def create_generator(self, limit: int | None = None) -> EventGenerator:
...
-async def _status_generator(
- instance: InstanceInfo,
- path: str, params: Params = None) -> EventGenerator:
- next_path = path
- while next_path:
- response = await request("GET", next_path, params=params)
- yield [StatusEvent(instance, from_dict(Status, s)) for s in response.json()]
- next_path = _get_next_path(response.headers)
+class StatusTimeline(Timeline):
+ """
+ StatusTimeline is the base class for timelines which only return statuses.
+ """
+
+ def __init__(self, name: str, instance: InstanceInfo):
+ super().__init__(name, instance)
+
+ async def status_generator(
+ self,
+ path: str,
+ params: Params = None,
+ limit: int | None = None) -> EventGenerator:
+ async for events in fetch_timeline(self.instance, path, params, limit):
+ yield [StatusEvent(self.instance, from_dict(Status, s)) for s in events]
-class HomeTimeline(Timeline):
+
+class HomeTimeline(StatusTimeline):
"""
HomeTimeline loads events from the user's home timeline.
This timeline only ever returns events of type StatusEvent.
@@ -67,23 +93,45 @@ class HomeTimeline(Timeline):
def __init__(self, instance: InstanceInfo):
super().__init__("Home", instance)
- def create_generator(self, limit: int = 40):
- return _status_generator(self.instance, "/api/v1/timelines/home", {"limit": limit})
+ def create_generator(self, limit: int | None = None):
+ return self.status_generator("/api/v1/timelines/home", limit=limit)
+
+
+class PublicTimeline(StatusTimeline):
+ """PublicTimeline loads events from the public timeline."""
+ def __init__(self, name: str, instance: InstanceInfo, local: bool):
+ super().__init__(name, instance)
+ self.local = local
+ def public_timeline_generator(self, limit: int | None = None):
+ return self.status_generator("/api/v1/timelines/public", {"local": self.local}, limit)
-class LocalTimeline(Timeline):
+
+class LocalTimeline(PublicTimeline):
"""
LocalTimeline loads events from the user's local instance timeline.
This timeline only ever returns events of type StatusEvent.
"""
def __init__(self, instance: InstanceInfo):
- super().__init__("Local", instance)
+ super().__init__("Local", instance, True)
- def create_generator(self, limit: int = 40):
- return public_timeline_generator(self.instance, local=True, limit=limit)
+ def create_generator(self, limit: int | None = None):
+ return self.public_timeline_generator(limit=limit)
-class AccountTimeline(Timeline):
+class FederatedTimeline(PublicTimeline):
+ """
+ FederatedTimeline loads events from the user's federated timeline.
+ This timeline only ever returns events of type StatusEvent.
+ """
+ def __init__(self, instance: InstanceInfo):
+ super().__init__("Federated", instance, False)
+
+ def create_generator(self, limit: int | None = None):
+ return self.public_timeline_generator(limit=limit)
+
+
+class AccountTimeline(StatusTimeline):
"""
AccountTimeline loads events from the given account's timeline.
This timeline only ever returns events of type StatusEvent.
@@ -109,26 +157,13 @@ class AccountTimeline(Timeline):
account = await get_account_by_name(account_name)
return AccountTimeline(instance, account_name, account.id, replies, reblogs)
- def create_generator(self, limit: int = 40):
+ def create_generator(self, limit: int | None = None):
path = f"/api/v1/accounts/{self.account_id}/statuses"
params = {
- "limit": limit,
"exclude_replies": not self.replies,
"exclude_reblogs": not self.reblogs
}
- return _status_generator(self.instance, path, params)
-
-
-class FederatedTimeline(Timeline):
- """
- FederatedTimeline loads events from the user's federated timeline.
- This timeline only ever returns events of type StatusEvent.
- """
- def __init__(self, instance: InstanceInfo):
- super().__init__("Federated", instance)
-
- def create_generator(self, limit: int = 40):
- return public_timeline_generator(self.instance, local=False, limit=limit)
+ return self.status_generator(path, params, limit)
class NotificationTimeline(Timeline):
@@ -154,23 +189,24 @@ class NotificationTimeline(Timeline):
case _:
return None
- async def _notification_generator(self, params: Params = None) -> EventGenerator:
- path = "/api/v1/notifications"
- next_path = path
+ async def notification_generator(
+ self,
+ path: str,
+ params: Params = None,
+ limit: int | None = None) -> EventGenerator:
- while next_path:
- response = await request("GET", next_path, params=params)
- yield [e for e in map(self.make_notification_event, response.json()) if e is not None]
- next_path = _get_next_path(response.headers)
+ path = "/api/v1/notifications"
+ async for events in fetch_timeline(self.instance, path, params, limit):
+ yield [e for e in map(self.make_notification_event, events) if e is not None]
- def create_generator(self, limit: int = 40):
+ def create_generator(self, limit: int | None = None):
# TODO: not included: follow_request, poll, update, admin.sign_up, admin.report
types = ["mention", "status", "reblog", "favourite", "follow"]
- params = {"types[]": types, "limit": limit}
- return self._notification_generator(params)
+ params = {"types[]": types}
+ return self.notification_generator(params, limit)
-class TagTimeline(Timeline):
+class TagTimeline(StatusTimeline):
"""
TagTimeline loads events from the given hashtag.
This timeline only ever returns events of type StatusEvent.
@@ -179,7 +215,7 @@ class TagTimeline(Timeline):
self.local = local
# Normalise the hashtag to not begin with a hash
- if hashtag[0] == '#':
+ if hashtag[0:1] == '#':
hashtag = hashtag[1:]
if len(hashtag) == 0:
@@ -189,7 +225,8 @@ class TagTimeline(Timeline):
super().__init__(f"#{self.hashtag}", instance)
def create_generator(self, limit: int = 40):
- return tag_timeline_generator(self.instance, self.hashtag, self.local, limit)
+ path = f"/api/v1/timelines/tag/{quote(self.hashtag)}"
+ return self.status_generator(path, limit=limit)
class ContextTimeline(Timeline):
@@ -201,38 +238,22 @@ class ContextTimeline(Timeline):
super().__init__("Thread", instance)
self._status = status
- def create_generator(self, limit: int = 40):
- async def _context_timeline_generator(status: Status, limit: int = 40):
- response = await statuses.context(status.original.id)
- data = response.json()
- ancestors = [from_dict(Status, s) for s in data["ancestors"]]
- descendants = [from_dict(Status, s) for s in data["descendants"]]
- all_statuses = ancestors + [status] + descendants
- yield [StatusEvent(self.instance, s) for s in all_statuses]
-
- return _context_timeline_generator(self._status, limit)
-
+ async def context_timeline_generator(self, status: Status, limit: int | None = None):
+ response = await statuses.context(status.original.id)
+ data = response.json()
+ ancestors = [from_dict(Status, s) for s in data["ancestors"]]
+ descendants = [from_dict(Status, s) for s in data["descendants"]]
+ all_statuses = ancestors + [status] + descendants
+ yield [StatusEvent(self.instance, s) for s in all_statuses]
-def public_timeline_generator(instance: InstanceInfo, local: bool = False, limit: int = 40):
- path = "/api/v1/timelines/public"
- params = {"local": str_bool(local), "limit": limit}
- return _status_generator(instance, path, params)
+ def create_generator(self, limit: int | None = None):
+ return self.context_timeline_generator(self._status, limit)
-def tag_timeline_generator(
- instance: InstanceInfo,
- hashtag: str,
- local: bool = False,
- limit: int = 40):
- path = f"/api/v1/timelines/tag/{quote(hashtag)}"
- params = {"local": str_bool(local), "limit": limit}
- return _status_generator(instance, path, params)
-
-
-def bookmark_timeline_generator(instance: InstanceInfo, limit: int = 40):
- path = "/api/v1/bookmarks"
- params = {"limit": limit}
- return _status_generator(instance, path, params)
+# def bookmark_timeline_generator(instance: InstanceInfo, limit: int = 40):
+# path = "/api/v1/bookmarks"
+# params = {"limit": limit}
+# return _status_generator(instance, path, params)
# def conversation_timeline_generator(limit: int = 40):
@@ -266,11 +287,13 @@ def bookmark_timeline_generator(instance: InstanceInfo, limit: int = 40):
# next_path = _get_next_path(response.headers)
-def _get_next_path(headers: Headers) -> str | None:
- """Given timeline response headers, returns the path to the next batch"""
- links = headers.get("Link", "")
- matches = re.match(r'<([^>]+)>; rel="next"', links)
- if matches:
- parsed = urlparse(matches.group(1))
- return "?".join([parsed.path, parsed.query])
- return None
+# def anon_public_timeline_generator(instance, local: bool = False, limit: int = 40):
+# path = "/api/v1/timelines/public"
+# params = {"local": str_bool(local), "limit": limit}
+# return _anon_timeline_generator(instance, path, params)
+
+
+# def anon_tag_timeline_generator(instance, hashtag, local: bool = False, limit: int = 40):
+# path = f"/api/v1/timelines/tag/{quote(hashtag)}"
+# params = {"local": str_bool(local), "limit": limit}
+# return _anon_timeline_generator(instance, path, params)
--
2.43.0
Applied. Thanks. -- Ivan