coop/rose-ash
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Decouple all cross-app service calls to HTTP endpoints
All checks were successful
Build and Deploy / build-and-deploy (push) Successful in 3m0s
Details

Browse Source 
Replace every direct cross-app services.* call with HTTP-based
communication: call_action() for writes, fetch_data() for reads.
Each app now registers only its own domain service.

Infrastructure:
- shared/infrastructure/actions.py — POST client for /internal/actions/
- shared/infrastructure/data_client.py — GET client for /internal/data/
- shared/contracts/dtos.py — dto_to_dict/dto_from_dict serialization

Action endpoints (writes):
- events: 8 handlers (ticket adjust, claim/confirm, toggle, adopt)
- market: 2 handlers (create/soft-delete marketplace)
- cart: 1 handler (adopt cart for user)

Data endpoints (reads):
- blog: 4 (post-by-slug/id, posts-by-ids, search-posts)
- events: 10 (pending entries/tickets, entries/tickets for page/order,
  entry-ids, associated-entries, calendars, visible-entries-for-period)
- market: 1 (marketplaces-for-container)
- cart: 1 (cart-summary)

Service registration cleanup:
- blog→blog+federation, events→calendar+federation,
  market→market+federation, cart→cart only,
  federation→federation only, account→nothing
- Stubs reduced to minimal StubFederationService

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
giles
2026-02-25 03:01:38 +00:00
parent 5dafbdbda9
commit 3b707ec8a0
55 changed files with 1210 additions and 581 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
shared/contracts/dtos.py
View File

@@ -5,11 +5,74 @@ see ORM model instances from another domain — only these DTOs.
"""
from __future__ import annotations
import dataclasses
import typing
from dataclasses import dataclass, field
from datetime import datetime
from decimal import Decimal
# ---------------------------------------------------------------------------
# Serialization helpers for JSON transport over internal data endpoints
# ---------------------------------------------------------------------------
def _serialize_value(v):
"""Convert a single value to a JSON-safe type."""
if isinstance(v, datetime):
return v.isoformat()
if isinstance(v, Decimal):
return str(v)
if isinstance(v, set):
return list(v)
if dataclasses.is_dataclass(v) and not isinstance(v, type):
return dto_to_dict(v)
if isinstance(v, list):
return [_serialize_value(item) for item in v]
return v
def dto_to_dict(obj) -> dict:
"""Convert a frozen DTO dataclass to a JSON-serialisable dict."""
return {k: _serialize_value(v) for k, v in dataclasses.asdict(obj).items()}
def _unwrap_optional(hint):
"""Unwrap Optional[X] / X | None to return X."""
args = getattr(hint, "__args__", ())
if args:
real = [a for a in args if a is not type(None)]
if real:
return real[0]
return hint
def dto_from_dict(cls, data: dict):
"""Construct a DTO from a dict, coercing dates and Decimals.
Uses ``typing.get_type_hints()`` to resolve forward-ref annotations
(from ``from __future__ import annotations``).
"""
if not data:
return None
try:
hints = typing.get_type_hints(cls)
except Exception:
hints = {}
kwargs = {}
for f in dataclasses.fields(cls):
if f.name not in data:
continue
val = data[f.name]
if val is not None and f.name in hints:
hint = _unwrap_optional(hints[f.name])
if hint is datetime and isinstance(val, str):
val = datetime.fromisoformat(val)
elif hint is Decimal:
val = Decimal(str(val))
kwargs[f.name] = val
return cls(**kwargs)
# ---------------------------------------------------------------------------
# Blog domain
# ---------------------------------------------------------------------------

22
shared/events/handlers/login_handlers.py
View File

@@ -1,23 +1,31 @@
from __future__ import annotations
import logging
from sqlalchemy.ext.asyncio import AsyncSession
from shared.events import register_activity_handler
from shared.infrastructure.actions import call_action, ActionError
from shared.models.federation import APActivity
from shared.services.registry import services
log = logging.getLogger(__name__)
async def on_user_logged_in(activity: APActivity, session: AsyncSession) -> None:
data = activity.object_data
user_id = data["user_id"]
session_id = data["session_id"]
payload = {"user_id": user_id, "session_id": session_id}
if services.has("cart"):
await services.cart.adopt_cart_for_user(session, user_id, session_id)
if services.has("calendar"):
await services.calendar.adopt_entries_for_user(session, user_id, session_id)
await services.calendar.adopt_tickets_for_user(session, user_id, session_id)
for app, action in [
("cart", "adopt-cart-for-user"),
("events", "adopt-entries-for-user"),
("events", "adopt-tickets-for-user"),
]:
try:
await call_action(app, action, payload=payload)
except ActionError:
log.warning("Failed: %s/%s for user %s", app, action, user_id)
register_activity_handler("rose:Login", on_user_logged_in)

89
shared/infrastructure/actions.py Normal file
View File

@@ -0,0 +1,89 @@
"""Internal action client for cross-app write operations.
Each coop app exposes JSON action endpoints at ``/internal/actions/{name}``.
This module provides helpers to call those endpoints so that callers don't
need direct access to another app's DB session or service layer.
Failures raise ``ActionError`` so callers can handle or propagate them.
"""
from __future__ import annotations
import logging
import os
import httpx
log = logging.getLogger(__name__)
# Re-usable async client (created lazily, one per process)
_client: httpx.AsyncClient | None = None
# Default request timeout (seconds) — longer than fragments since these are writes
_DEFAULT_TIMEOUT = 5.0
# Header sent on every action request so providers can gate access.
ACTION_HEADER = "X-Internal-Action"
class ActionError(Exception):
"""Raised when an internal action call fails."""
def __init__(self, message: str, status_code: int = 500, detail: dict | None = None):
super().__init__(message)
self.status_code = status_code
self.detail = detail
def _get_client() -> httpx.AsyncClient:
global _client
if _client is None or _client.is_closed:
_client = httpx.AsyncClient(
timeout=httpx.Timeout(_DEFAULT_TIMEOUT),
follow_redirects=False,
)
return _client
def _internal_url(app_name: str) -> str:
"""Resolve the Docker-internal base URL for *app_name*."""
env_key = f"INTERNAL_URL_{app_name.upper()}"
return os.getenv(env_key, f"http://{app_name}:8000").rstrip("/")
async def call_action(
app_name: str,
action_name: str,
*,
payload: dict | None = None,
timeout: float = _DEFAULT_TIMEOUT,
) -> dict:
"""POST JSON to ``{INTERNAL_URL_APP}/internal/actions/{action_name}``.
Returns the parsed JSON response on 2xx.
Raises ``ActionError`` on network errors or non-2xx responses.
"""
base = _internal_url(app_name)
url = f"{base}/internal/actions/{action_name}"
try:
resp = await _get_client().post(
url,
json=payload or {},
headers={ACTION_HEADER: "1"},
timeout=timeout,
)
if 200 <= resp.status_code < 300:
return resp.json()
msg = f"Action {app_name}/{action_name} returned {resp.status_code}"
detail = None
try:
detail = resp.json()
except Exception:
pass
log.error(msg)
raise ActionError(msg, status_code=resp.status_code, detail=detail)
except ActionError:
raise
except Exception as exc:
msg = f"Action {app_name}/{action_name} failed: {exc}"
log.error(msg)
raise ActionError(msg) from exc

91
shared/infrastructure/data_client.py Normal file
View File

@@ -0,0 +1,91 @@
"""Internal data client for cross-app read operations.
Each coop app exposes JSON data endpoints at ``/internal/data/{query}``.
This module provides helpers to fetch that data so that callers don't
need direct access to another app's DB session or service layer.
Same pattern as the fragment client but returns parsed JSON instead of HTML.
"""
from __future__ import annotations
import logging
import os
import httpx
log = logging.getLogger(__name__)
# Re-usable async client (created lazily, one per process)
_client: httpx.AsyncClient | None = None
# Default request timeout (seconds)
_DEFAULT_TIMEOUT = 3.0
# Header sent on every data request so providers can gate access.
DATA_HEADER = "X-Internal-Data"
class DataError(Exception):
"""Raised when an internal data fetch fails."""
def __init__(self, message: str, status_code: int = 500):
super().__init__(message)
self.status_code = status_code
def _get_client() -> httpx.AsyncClient:
global _client
if _client is None or _client.is_closed:
_client = httpx.AsyncClient(
timeout=httpx.Timeout(_DEFAULT_TIMEOUT),
follow_redirects=False,
)
return _client
def _internal_url(app_name: str) -> str:
"""Resolve the Docker-internal base URL for *app_name*."""
env_key = f"INTERNAL_URL_{app_name.upper()}"
return os.getenv(env_key, f"http://{app_name}:8000").rstrip("/")
async def fetch_data(
app_name: str,
query_name: str,
*,
params: dict | None = None,
timeout: float = _DEFAULT_TIMEOUT,
required: bool = True,
) -> dict | list | None:
"""GET JSON from ``{INTERNAL_URL_APP}/internal/data/{query_name}``.
Returns parsed JSON (dict or list) on success.
When *required* is True (default), raises ``DataError`` on failure.
When *required* is False, returns None on failure.
"""
base = _internal_url(app_name)
url = f"{base}/internal/data/{query_name}"
try:
resp = await _get_client().get(
url,
params=params,
headers={DATA_HEADER: "1"},
timeout=timeout,
)
if resp.status_code == 200:
return resp.json()
msg = f"Data {app_name}/{query_name} returned {resp.status_code}"
if required:
log.error(msg)
raise DataError(msg, status_code=resp.status_code)
log.warning(msg)
return None
except DataError:
raise
except Exception as exc:
msg = f"Data {app_name}/{query_name} failed: {exc}"
if required:
log.error(msg)
raise DataError(msg) from exc
log.warning(msg)
return None

57
shared/services/cart_impl.py
View File

@@ -13,7 +13,6 @@ from sqlalchemy.orm import selectinload
from shared.models.market import CartItem
from shared.models.market_place import MarketPlace
from shared.models.calendars import CalendarEntry, Calendar
from shared.contracts.dtos import CartItemDTO, CartSummaryDTO
@@ -39,13 +38,16 @@ class SqlCartService:
page_slug: str | None = None,
) -> CartSummaryDTO:
"""Build a lightweight cart summary for the current identity."""
# Resolve page filter
from shared.infrastructure.data_client import fetch_data
from shared.contracts.dtos import CalendarEntryDTO, TicketDTO, dto_from_dict
# Resolve page filter via blog data endpoint
page_post_id: int | None = None
if page_slug:
from shared.services.registry import services
post = await services.blog.get_post_by_slug(session, page_slug)
if post and post.is_page:
page_post_id = post.id
post = await fetch_data("blog", "post-by-slug",
params={"slug": page_slug}, required=False)
if post and post.get("is_page"):
page_post_id = post["id"]
# --- product cart ---
cart_q = select(CartItem).where(CartItem.deleted_at.is_(None))
@@ -75,37 +77,34 @@ class SqlCartService:
if ci.product and (ci.product.special_price or ci.product.regular_price)
)
# --- calendar entries ---
from shared.services.registry import services
# --- calendar entries via events data endpoint ---
cal_params: dict = {}
if user_id is not None:
cal_params["user_id"] = user_id
if session_id is not None:
cal_params["session_id"] = session_id
if page_post_id is not None:
cal_entries = await services.calendar.entries_for_page(
session, page_post_id,
user_id=user_id,
session_id=session_id,
)
cal_params["page_id"] = page_post_id
raw_entries = await fetch_data("events", "entries-for-page",
params=cal_params, required=False) or []
else:
cal_entries = await services.calendar.pending_entries(
session,
user_id=user_id,
session_id=session_id,
)
raw_entries = await fetch_data("events", "pending-entries",
params=cal_params, required=False) or []
cal_entries = [dto_from_dict(CalendarEntryDTO, e) for e in raw_entries]
calendar_count = len(cal_entries)
calendar_total = sum(Decimal(str(e.cost or 0)) for e in cal_entries if e.cost is not None)
# --- tickets ---
# --- tickets via events data endpoint ---
if page_post_id is not None:
tickets = await services.calendar.tickets_for_page(
session, page_post_id,
user_id=user_id,
session_id=session_id,
)
raw_tickets = await fetch_data("events", "tickets-for-page",
params=cal_params, required=False) or []
else:
tickets = await services.calendar.pending_tickets(
session,
user_id=user_id,
session_id=session_id,
)
tk_params = {k: v for k, v in cal_params.items() if k != "page_id"}
raw_tickets = await fetch_data("events", "pending-tickets",
params=tk_params, required=False) or []
tickets = [dto_from_dict(TicketDTO, t) for t in raw_tickets]
ticket_count = len(tickets)
ticket_total = sum(Decimal(str(t.price or 0)) for t in tickets)

13
shared/services/registry.py
View File

@@ -1,18 +1,17 @@
"""Typed singleton registry for domain services.
Each app registers ONLY its own domain service. Cross-app calls go
over HTTP via ``call_action()`` (writes) and ``fetch_data()`` (reads).
Usage::
from shared.services.registry import services
# Register at app startup
# Register at app startup (own domain only)
services.blog = SqlBlogService()
# Query anywhere
if services.has("calendar"):
entries = await services.calendar.pending_entries(session, ...)
# Or use stubs for absent domains
summary = await services.cart.cart_summary(session, ...)
# Use locally within the owning app
post = await services.blog.get_post_by_slug(session, slug)
"""
from __future__ import annotations

294
shared/services/stubs.py
View File

@@ -1,204 +1,15 @@
"""No-op stub services for absent domains.
"""No-op stub services.
When an app starts without a particular domain, it registers the stub
so that ``services.X.method()`` returns empty/None rather than crashing.
Cross-app calls now go over HTTP via call_action() / fetch_data().
Stubs are no longer needed for the 4 main domains (blog, calendar,
market, cart). Only StubFederationService remains as a safety net
for apps that conditionally load AP infrastructure.
"""
from __future__ import annotations
from decimal import Decimal
from sqlalchemy.ext.asyncio import AsyncSession
from shared.contracts.dtos import (
PostDTO,
CalendarDTO,
CalendarEntryDTO,
TicketDTO,
MarketPlaceDTO,
ProductDTO,
CartItemDTO,
CartSummaryDTO,
ActorProfileDTO,
APActivityDTO,
APFollowerDTO,
)
class StubBlogService:
async def get_post_by_slug(self, session: AsyncSession, slug: str) -> PostDTO | None:
return None
async def get_post_by_id(self, session: AsyncSession, id: int) -> PostDTO | None:
return None
async def get_posts_by_ids(self, session: AsyncSession, ids: list[int]) -> list[PostDTO]:
return []
async def search_posts(self, session, query, page=1, per_page=10):
return [], 0
class StubCalendarService:
async def calendars_for_container(
self, session: AsyncSession, container_type: str, container_id: int,
) -> list[CalendarDTO]:
return []
async def pending_entries(
self, session: AsyncSession, *, user_id: int | None, session_id: str | None,
) -> list[CalendarEntryDTO]:
return []
async def entries_for_page(
self, session: AsyncSession, page_id: int, *, user_id: int | None, session_id: str | None,
) -> list[CalendarEntryDTO]:
return []
async def entry_by_id(self, session: AsyncSession, entry_id: int) -> CalendarEntryDTO | None:
return None
async def associated_entries(
self, session: AsyncSession, content_type: str, content_id: int, page: int,
) -> tuple[list[CalendarEntryDTO], bool]:
return [], False
async def toggle_entry_post(
self, session: AsyncSession, entry_id: int, content_type: str, content_id: int,
) -> bool:
return False
async def adopt_entries_for_user(
self, session: AsyncSession, user_id: int, session_id: str,
) -> None:
pass
async def claim_entries_for_order(
self, session: AsyncSession, order_id: int, user_id: int | None,
session_id: str | None, page_post_id: int | None,
) -> None:
pass
async def confirm_entries_for_order(
self, session: AsyncSession, order_id: int, user_id: int | None,
session_id: str | None,
) -> None:
pass
async def get_entries_for_order(
self, session: AsyncSession, order_id: int,
) -> list[CalendarEntryDTO]:
return []
async def user_tickets(
self, session: AsyncSession, *, user_id: int,
) -> list[TicketDTO]:
return []
async def user_bookings(
self, session: AsyncSession, *, user_id: int,
) -> list[CalendarEntryDTO]:
return []
async def confirmed_entries_for_posts(
self, session: AsyncSession, post_ids: list[int],
) -> dict[int, list[CalendarEntryDTO]]:
return {}
async def pending_tickets(
self, session: AsyncSession, *, user_id: int | None, session_id: str | None,
) -> list[TicketDTO]:
return []
async def tickets_for_page(
self, session: AsyncSession, page_id: int, *, user_id: int | None, session_id: str | None,
) -> list[TicketDTO]:
return []
async def claim_tickets_for_order(
self, session: AsyncSession, order_id: int, user_id: int | None,
session_id: str | None, page_post_id: int | None,
) -> None:
pass
async def confirm_tickets_for_order(
self, session: AsyncSession, order_id: int,
) -> None:
pass
async def get_tickets_for_order(
self, session: AsyncSession, order_id: int,
) -> list[TicketDTO]:
return []
async def adopt_tickets_for_user(
self, session: AsyncSession, user_id: int, session_id: str,
) -> None:
pass
async def adjust_ticket_quantity(
self, session, entry_id, count, *, user_id, session_id, ticket_type_id=None,
) -> int:
return 0
async def upcoming_entries_for_container(self, session, container_type, container_id, *, page=1, per_page=20):
return [], False
async def entry_ids_for_content(self, session, content_type, content_id):
return set()
async def visible_entries_for_period(self, session, calendar_id, period_start, period_end, *, user_id, is_admin, session_id):
return []
class StubMarketService:
async def marketplaces_for_container(
self, session: AsyncSession, container_type: str, container_id: int,
) -> list[MarketPlaceDTO]:
return []
async def list_marketplaces(
self, session: AsyncSession,
container_type: str | None = None, container_id: int | None = None,
*, page: int = 1, per_page: int = 20,
) -> tuple[list[MarketPlaceDTO], bool]:
return [], False
async def product_by_id(self, session: AsyncSession, product_id: int) -> ProductDTO | None:
return None
async def create_marketplace(
self, session: AsyncSession, container_type: str, container_id: int,
name: str, slug: str,
) -> MarketPlaceDTO:
raise RuntimeError("MarketService not available")
async def soft_delete_marketplace(
self, session: AsyncSession, container_type: str, container_id: int,
slug: str,
) -> bool:
return False
class StubCartService:
async def cart_summary(
self, session: AsyncSession, *, user_id: int | None, session_id: str | None,
page_slug: str | None = None,
) -> CartSummaryDTO:
return CartSummaryDTO()
async def cart_items(
self, session: AsyncSession, *, user_id: int | None, session_id: str | None,
) -> list[CartItemDTO]:
return []
async def adopt_cart_for_user(
self, session: AsyncSession, user_id: int, session_id: str,
) -> None:
pass
class StubFederationService:
"""No-op federation stub for apps that don't own federation."""
"""No-op federation stub for apps that don't load AP."""
async def get_actor_by_username(self, session, username):
return None
@@ -206,109 +17,16 @@ class StubFederationService:
async def get_actor_by_user_id(self, session, user_id):
return None
async def create_actor(self, session, user_id, preferred_username,
display_name=None, summary=None):
raise RuntimeError("FederationService not available")
async def username_available(self, session, username):
return False
async def publish_activity(self, session, *, actor_user_id, activity_type,
object_type, object_data, source_type=None,
source_id=None):
return None
async def get_activity(self, session, activity_id):
return None
async def get_outbox(self, session, username, page=1, per_page=20, origin_app=None):
return [], 0
async def get_activity_for_source(self, session, source_type, source_id):
return None
async def count_activities_for_source(self, session, source_type, source_id, *, activity_type):
return 0
async def get_followers(self, session, username, app_domain=None):
return []
async def add_follower(self, session, username, follower_acct, follower_inbox,
follower_actor_url, follower_public_key=None,
app_domain="federation"):
raise RuntimeError("FederationService not available")
async def remove_follower(self, session, username, follower_acct, app_domain="federation"):
return False
async def get_or_fetch_remote_actor(self, session, actor_url):
return None
async def search_remote_actor(self, session, acct):
return None
async def search_actors(self, session, query, page=1, limit=20):
return [], 0
async def send_follow(self, session, local_username, remote_actor_url):
raise RuntimeError("FederationService not available")
async def get_following(self, session, username, page=1, per_page=20):
return [], 0
async def get_followers_paginated(self, session, username, page=1, per_page=20):
return [], 0
async def accept_follow_response(self, session, local_username, remote_actor_url):
pass
async def unfollow(self, session, local_username, remote_actor_url):
pass
async def ingest_remote_post(self, session, remote_actor_id, activity_json, object_json):
pass
async def delete_remote_post(self, session, object_id):
pass
async def get_remote_post(self, session, object_id):
return None
async def get_home_timeline(self, session, actor_profile_id, before=None, limit=20):
return []
async def get_public_timeline(self, session, before=None, limit=20):
return []
async def get_actor_timeline(self, session, remote_actor_id, before=None, limit=20):
return []
async def create_local_post(self, session, actor_profile_id, content, visibility="public", in_reply_to=None):
raise RuntimeError("FederationService not available")
async def delete_local_post(self, session, actor_profile_id, post_id):
raise RuntimeError("FederationService not available")
async def like_post(self, session, actor_profile_id, object_id, author_inbox):
pass
async def unlike_post(self, session, actor_profile_id, object_id, author_inbox):
pass
async def boost_post(self, session, actor_profile_id, object_id, author_inbox):
pass
async def unboost_post(self, session, actor_profile_id, object_id, author_inbox):
pass
async def get_notifications(self, session, actor_profile_id, before=None, limit=20):
return []
async def unread_notification_count(self, session, actor_profile_id):
return 0
async def mark_notifications_read(self, session, actor_profile_id):
pass
async def get_stats(self, session):
return {"actors": 0, "activities": 0, "followers": 0}