art-dag/mono
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Eliminate Python page helpers from account, federation, and cart

Browse Source 
All three services now fetch page data via (service ...) IO primitives
in .sx defpages instead of Python middleman functions.

- Account: newsletters-data → AccountPageService.newsletters_data
- Federation: 8 page helpers → FederationPageService methods
  (timeline, compose, search, following, followers, notifications)
- Cart: 4 page helpers → CartPageService methods
  (overview, page-cart, admin, payments)
- Serializers moved to service modules, thin delegates kept for routes
- ~520 lines of Python page helpers removed

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
giles
2026-03-04 02:01:50 +00:00
parent fb8f115acb
commit 418ac9424f
12 changed files with 520 additions and 521 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
federation/services/__init__.py
View File

@@ -13,3 +13,6 @@ def register_domain_services() -> None:
from shared.services.federation_impl import SqlFederationService
services.federation = SqlFederationService()
from .federation_page import FederationPageService
services.register("federation_page", FederationPageService())

205
federation/services/federation_page.py Normal file
View File

@@ -0,0 +1,205 @@
"""Federation page data service — provides serialized dicts for .sx defpages."""
from __future__ import annotations
def _serialize_actor(actor) -> dict | None:
if not actor:
return None
return {
"id": actor.id,
"preferred_username": actor.preferred_username,
"display_name": getattr(actor, "display_name", None),
"icon_url": getattr(actor, "icon_url", None),
"summary": getattr(actor, "summary", None),
"actor_url": getattr(actor, "actor_url", ""),
"domain": getattr(actor, "domain", ""),
}
def _serialize_timeline_item(item) -> dict:
published = getattr(item, "published", None)
return {
"object_id": getattr(item, "object_id", "") or "",
"author_inbox": getattr(item, "author_inbox", "") or "",
"actor_icon": getattr(item, "actor_icon", None),
"actor_name": getattr(item, "actor_name", "?"),
"actor_username": getattr(item, "actor_username", ""),
"actor_domain": getattr(item, "actor_domain", ""),
"content": getattr(item, "content", ""),
"summary": getattr(item, "summary", None),
"published": published.strftime("%b %d, %H:%M") if published else "",
"before_cursor": published.isoformat() if published else "",
"url": getattr(item, "url", None),
"post_type": getattr(item, "post_type", ""),
"boosted_by": getattr(item, "boosted_by", None),
"like_count": getattr(item, "like_count", 0) or 0,
"boost_count": getattr(item, "boost_count", 0) or 0,
"liked_by_me": getattr(item, "liked_by_me", False),
"boosted_by_me": getattr(item, "boosted_by_me", False),
}
def _serialize_remote_actor(a) -> dict:
return {
"id": getattr(a, "id", None),
"display_name": getattr(a, "display_name", None) or getattr(a, "preferred_username", ""),
"preferred_username": getattr(a, "preferred_username", ""),
"domain": getattr(a, "domain", ""),
"icon_url": getattr(a, "icon_url", None),
"actor_url": getattr(a, "actor_url", ""),
"summary": getattr(a, "summary", None),
}
def _get_actor():
from quart import g
return getattr(g, "_social_actor", None)
def _require_actor():
from quart import abort
actor = _get_actor()
if not actor:
abort(403, "You need to choose a federation username first")
return actor
class FederationPageService:
"""Service for federation page data, callable via (service "federation-page" ...)."""
async def home_timeline_data(self, session, **kw):
actor = _require_actor()
from shared.services.registry import services
items = await services.federation.get_home_timeline(session, actor.id)
return {
"items": [_serialize_timeline_item(i) for i in items],
"timeline_type": "home",
"actor": _serialize_actor(actor),
}
async def public_timeline_data(self, session, **kw):
actor = _get_actor()
from shared.services.registry import services
items = await services.federation.get_public_timeline(session)
return {
"items": [_serialize_timeline_item(i) for i in items],
"timeline_type": "public",
"actor": _serialize_actor(actor),
}
async def compose_data(self, session, **kw):
from quart import request
_require_actor()
reply_to = request.args.get("reply_to")
return {"reply_to": reply_to or None}
async def search_data(self, session, **kw):
from quart import request
actor = _get_actor()
from shared.services.registry import services
query = request.args.get("q", "").strip()
actors_list = []
total = 0
followed_urls: list[str] = []
if query:
actors_list, total = await services.federation.search_actors(session, query)
if actor:
following, _ = await services.federation.get_following(
session, actor.preferred_username, page=1, per_page=1000,
)
followed_urls = [a.actor_url for a in following]
return {
"query": query,
"actors": [_serialize_remote_actor(a) for a in actors_list],
"total": total,
"followed_urls": followed_urls,
"actor": _serialize_actor(actor),
}
async def following_data(self, session, **kw):
actor = _require_actor()
from shared.services.registry import services
actors_list, total = await services.federation.get_following(
session, actor.preferred_username,
)
return {
"actors": [_serialize_remote_actor(a) for a in actors_list],
"total": total,
"actor": _serialize_actor(actor),
}
async def followers_data(self, session, **kw):
actor = _require_actor()
from shared.services.registry import services
actors_list, total = await services.federation.get_followers_paginated(
session, actor.preferred_username,
)
following, _ = await services.federation.get_following(
session, actor.preferred_username, page=1, per_page=1000,
)
followed_urls = [a.actor_url for a in following]
return {
"actors": [_serialize_remote_actor(a) for a in actors_list],
"total": total,
"followed_urls": followed_urls,
"actor": _serialize_actor(actor),
}
async def actor_timeline_data(self, session, *, id=None, **kw):
from quart import abort
from sqlalchemy import select as sa_select
from shared.models.federation import RemoteActor
from shared.services.registry import services
from shared.services.federation_impl import _remote_actor_to_dto
actor = _get_actor()
actor_id = id
remote = (
await session.execute(
sa_select(RemoteActor).where(RemoteActor.id == actor_id)
)
).scalar_one_or_none()
if not remote:
abort(404)
remote_dto = _remote_actor_to_dto(remote)
items = await services.federation.get_actor_timeline(session, actor_id)
is_following = False
if actor:
from shared.models.federation import APFollowing
existing = (
await session.execute(
sa_select(APFollowing).where(
APFollowing.actor_profile_id == actor.id,
APFollowing.remote_actor_id == actor_id,
)
)
).scalar_one_or_none()
is_following = existing is not None
return {
"remote_actor": _serialize_remote_actor(remote_dto),
"items": [_serialize_timeline_item(i) for i in items],
"is_following": is_following,
"actor": _serialize_actor(actor),
}
async def notifications_data(self, session, **kw):
actor = _require_actor()
from shared.services.registry import services
items = await services.federation.get_notifications(session, actor.id)
await services.federation.mark_notifications_read(session, actor.id)
notif_dicts = []
for n in items:
created = getattr(n, "created_at", None)
notif_dicts.append({
"from_actor_name": getattr(n, "from_actor_name", "?"),
"from_actor_username": getattr(n, "from_actor_username", ""),
"from_actor_domain": getattr(n, "from_actor_domain", ""),
"from_actor_icon": getattr(n, "from_actor_icon", None),
"notification_type": getattr(n, "notification_type", ""),
"target_content_preview": getattr(n, "target_content_preview", None),
"created_at_formatted": created.strftime("%b %d, %H:%M") if created else "",
"read": getattr(n, "read", True),
"app_domain": getattr(n, "app_domain", ""),
})
return {"notifications": notif_dicts}

221
federation/sxc/pages/__init__.py
View File

@@ -1,13 +1,12 @@
"""Federation defpage setup — registers layouts, page helpers, and loads .sx pages."""
"""Federation defpage setup — registers layouts and loads .sx pages."""
from __future__ import annotations
from typing import Any
def setup_federation_pages() -> None:
"""Register federation-specific layouts, page helpers, and load page definitions."""
"""Register federation-specific layouts and load page definitions."""
_register_federation_layouts()
_register_federation_helpers()
_load_federation_page_files()
@@ -55,74 +54,25 @@ async def _social_oob(ctx: dict, **kw: Any) -> str:
# ---------------------------------------------------------------------------
# Page helpers
# Serializers and helpers — still used by layouts and route handlers
# ---------------------------------------------------------------------------
def _register_federation_helpers() -> None:
from shared.sx.pages import register_page_helpers
register_page_helpers("federation", {
"home-timeline-content": _h_home_timeline_content,
"public-timeline-content": _h_public_timeline_content,
"compose-content": _h_compose_content,
"search-content": _h_search_content,
"following-content": _h_following_content,
"followers-content": _h_followers_content,
"actor-timeline-content": _h_actor_timeline_content,
"notifications-content": _h_notifications_content,
})
def _serialize_actor(actor) -> dict | None:
"""Serialize an actor profile to a dict for sx defcomps."""
if not actor:
return None
return {
"id": actor.id,
"preferred_username": actor.preferred_username,
"display_name": getattr(actor, "display_name", None),
"icon_url": getattr(actor, "icon_url", None),
"summary": getattr(actor, "summary", None),
"actor_url": getattr(actor, "actor_url", ""),
"domain": getattr(actor, "domain", ""),
}
from services.federation_page import _serialize_actor as _impl
return _impl(actor)
def _serialize_timeline_item(item) -> dict:
"""Serialize a timeline item DTO to a dict for sx defcomps."""
published = getattr(item, "published", None)
return {
"object_id": getattr(item, "object_id", "") or "",
"author_inbox": getattr(item, "author_inbox", "") or "",
"actor_icon": getattr(item, "actor_icon", None),
"actor_name": getattr(item, "actor_name", "?"),
"actor_username": getattr(item, "actor_username", ""),
"actor_domain": getattr(item, "actor_domain", ""),
"content": getattr(item, "content", ""),
"summary": getattr(item, "summary", None),
"published": published.strftime("%b %d, %H:%M") if published else "",
"before_cursor": published.isoformat() if published else "",
"url": getattr(item, "url", None),
"post_type": getattr(item, "post_type", ""),
"boosted_by": getattr(item, "boosted_by", None),
"like_count": getattr(item, "like_count", 0) or 0,
"boost_count": getattr(item, "boost_count", 0) or 0,
"liked_by_me": getattr(item, "liked_by_me", False),
"boosted_by_me": getattr(item, "boosted_by_me", False),
}
from services.federation_page import _serialize_timeline_item as _impl
return _impl(item)
def _serialize_remote_actor(a) -> dict:
"""Serialize a remote actor DTO to a dict for sx defcomps."""
return {
"id": getattr(a, "id", None),
"display_name": getattr(a, "display_name", None) or getattr(a, "preferred_username", ""),
"preferred_username": getattr(a, "preferred_username", ""),
"domain": getattr(a, "domain", ""),
"icon_url": getattr(a, "icon_url", None),
"actor_url": getattr(a, "actor_url", ""),
"summary": getattr(a, "summary", None),
}
from services.federation_page import _serialize_remote_actor as _impl
return _impl(a)
async def _social_page(ctx: dict, actor, *, content: str,
@@ -155,156 +105,3 @@ def _require_actor():
if not actor:
abort(403, "You need to choose a federation username first")
return actor
async def _h_home_timeline_content(**kw):
from quart import g
from shared.services.registry import services
from shared.sx.helpers import render_to_sx
actor = _require_actor()
items = await services.federation.get_home_timeline(g.s, actor.id)
return await render_to_sx("federation-timeline-content",
items=[_serialize_timeline_item(i) for i in items],
timeline_type="home",
actor=_serialize_actor(actor))
async def _h_public_timeline_content(**kw):
from quart import g
from shared.services.registry import services
from shared.sx.helpers import render_to_sx
actor = _get_actor()
items = await services.federation.get_public_timeline(g.s)
return await render_to_sx("federation-timeline-content",
items=[_serialize_timeline_item(i) for i in items],
timeline_type="public",
actor=_serialize_actor(actor))
async def _h_compose_content(**kw):
from quart import request
from shared.sx.helpers import render_to_sx
_require_actor()
reply_to = request.args.get("reply_to")
return await render_to_sx("federation-compose-content",
reply_to=reply_to or None)
async def _h_search_content(**kw):
from quart import g, request
from shared.services.registry import services
from shared.sx.helpers import render_to_sx
actor = _get_actor()
query = request.args.get("q", "").strip()
actors_list = []
total = 0
followed_urls: set[str] = set()
if query:
actors_list, total = await services.federation.search_actors(g.s, query)
if actor:
following, _ = await services.federation.get_following(
g.s, actor.preferred_username, page=1, per_page=1000,
)
followed_urls = {a.actor_url for a in following}
return await render_to_sx("federation-search-content",
query=query,
actors=[_serialize_remote_actor(a) for a in actors_list],
total=total,
followed_urls=list(followed_urls),
actor=_serialize_actor(actor))
async def _h_following_content(**kw):
from quart import g
from shared.services.registry import services
from shared.sx.helpers import render_to_sx
actor = _require_actor()
actors_list, total = await services.federation.get_following(
g.s, actor.preferred_username,
)
return await render_to_sx("federation-following-content",
actors=[_serialize_remote_actor(a) for a in actors_list],
total=total,
actor=_serialize_actor(actor))
async def _h_followers_content(**kw):
from quart import g
from shared.services.registry import services
from shared.sx.helpers import render_to_sx
actor = _require_actor()
actors_list, total = await services.federation.get_followers_paginated(
g.s, actor.preferred_username,
)
following, _ = await services.federation.get_following(
g.s, actor.preferred_username, page=1, per_page=1000,
)
followed_urls = {a.actor_url for a in following}
return await render_to_sx("federation-followers-content",
actors=[_serialize_remote_actor(a) for a in actors_list],
total=total,
followed_urls=list(followed_urls),
actor=_serialize_actor(actor))
async def _h_actor_timeline_content(id=None, **kw):
from quart import g, abort
from shared.services.registry import services
from shared.sx.helpers import render_to_sx
actor = _get_actor()
actor_id = id
from shared.models.federation import RemoteActor
from sqlalchemy import select as sa_select
remote = (
await g.s.execute(
sa_select(RemoteActor).where(RemoteActor.id == actor_id)
)
).scalar_one_or_none()
if not remote:
abort(404)
from shared.services.federation_impl import _remote_actor_to_dto
remote_dto = _remote_actor_to_dto(remote)
items = await services.federation.get_actor_timeline(g.s, actor_id)
is_following = False
if actor:
from shared.models.federation import APFollowing
existing = (
await g.s.execute(
sa_select(APFollowing).where(
APFollowing.actor_profile_id == actor.id,
APFollowing.remote_actor_id == actor_id,
)
)
).scalar_one_or_none()
is_following = existing is not None
return await render_to_sx("federation-actor-timeline-content",
remote_actor=_serialize_remote_actor(remote_dto),
items=[_serialize_timeline_item(i) for i in items],
is_following=is_following,
actor=_serialize_actor(actor))
async def _h_notifications_content(**kw):
from quart import g
from shared.services.registry import services
from shared.sx.helpers import render_to_sx
actor = _require_actor()
items = await services.federation.get_notifications(g.s, actor.id)
await services.federation.mark_notifications_read(g.s, actor.id)
notif_dicts = []
for n in items:
created = getattr(n, "created_at", None)
notif_dicts.append({
"from_actor_name": getattr(n, "from_actor_name", "?"),
"from_actor_username": getattr(n, "from_actor_username", ""),
"from_actor_domain": getattr(n, "from_actor_domain", ""),
"from_actor_icon": getattr(n, "from_actor_icon", None),
"notification_type": getattr(n, "notification_type", ""),
"target_content_preview": getattr(n, "target_content_preview", None),
"created_at_formatted": created.strftime("%b %d, %H:%M") if created else "",
"read": getattr(n, "read", True),
"app_domain": getattr(n, "app_domain", ""),
})
return await render_to_sx("federation-notifications-content",
notifications=notif_dicts)

49
federation/sxc/pages/social.sx
View File

@@ -1,49 +1,82 @@
;; Federation social pages
;; All data fetching via (service ...) IO primitives, no Python helpers.
(defpage home-timeline
:path "/social/"
:auth :login
:layout :social
:content (home-timeline-content))
:data (service "federation-page" "home-timeline-data")
:content (~federation-timeline-content
:items items
:timeline-type timeline-type
:actor actor))
(defpage public-timeline
:path "/social/public"
:auth :public
:layout :social
:content (public-timeline-content))
:data (service "federation-page" "public-timeline-data")
:content (~federation-timeline-content
:items items
:timeline-type timeline-type
:actor actor))
(defpage compose-form
:path "/social/compose"
:auth :login
:layout :social
:content (compose-content))
:data (service "federation-page" "compose-data")
:content (~federation-compose-content
:reply-to reply-to))
(defpage search
:path "/social/search"
:auth :public
:layout :social
:content (search-content))
:data (service "federation-page" "search-data")
:content (~federation-search-content
:query query
:actors actors
:total total
:followed-urls followed-urls
:actor actor))
(defpage following-list
:path "/social/following"
:auth :login
:layout :social
:content (following-content))
:data (service "federation-page" "following-data")
:content (~federation-following-content
:actors actors
:total total
:actor actor))
(defpage followers-list
:path "/social/followers"
:auth :login
:layout :social
:content (followers-content))
:data (service "federation-page" "followers-data")
:content (~federation-followers-content
:actors actors
:total total
:followed-urls followed-urls
:actor actor))
(defpage actor-timeline
:path "/social/actor/<int:id>"
:auth :public
:layout :social
:content (actor-timeline-content id))
:data (service "federation-page" "actor-timeline-data" :id id)
:content (~federation-actor-timeline-content
:remote-actor remote-actor
:items items
:is-following is-following
:actor actor))
(defpage notifications
:path "/social/notifications"
:auth :login
:layout :social
:content (notifications-content))
:data (service "federation-page" "notifications-data")
:content (~federation-notifications-content
:notifications notifications))