giles
1f36987f77
All checks were successful
Build and Deploy / build-and-deploy (push) Successful in 2m5s
The inter-service data layer (fetch_data/call_action) was the least structured part of the codebase — Python _handlers dicts with ad-hoc param extraction scattered across 16 route files. This replaces them with declarative .sx query/action definitions that make the entire inter-service protocol self-describing and greppable. Infrastructure: - defquery/defaction special forms in the sx evaluator - Query/action registry with load, lookup, and schema introspection - Query executor using async_eval with I/O primitives - Blueprint factories (create_data_blueprint/create_action_blueprint) with sx-first dispatch and Python fallback - /internal/schema endpoint on every service - parse-datetime and split-ids primitives for type coercion Service extractions: - LikesService (toggle, is_liked, liked_slugs, liked_ids) - PageConfigService (ensure, get_by_container, get_by_id, get_batch, update) - RelationsService (wraps module-level functions) - AccountDataService (user_by_email, newsletters) - CartItemsService, MarketDataService (raw SQLAlchemy lookups) 50 of 54 handlers converted to sx, 4 Python fallbacks remain (ghost-sync/push-member, clear-cart-for-order, create-order). Net: -1,383 lines Python, +251 lines modified. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
125 lines
4.0 KiB
Python
125 lines
4.0 KiB
Python
"""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 (own domain only)
|
|
services.calendar = SqlCalendarService()
|
|
|
|
# Use locally within the owning app
|
|
cals = await services.calendar.calendars_for_container(session, "page", page_id)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from shared.contracts.protocols import (
|
|
CalendarService,
|
|
MarketService,
|
|
CartService,
|
|
FederationService,
|
|
)
|
|
from shared.contracts.likes import LikesService
|
|
|
|
|
|
class _ServiceRegistry:
|
|
"""Central registry holding one implementation per domain.
|
|
|
|
Properties return the registered implementation or raise
|
|
``RuntimeError`` if nothing is registered. Use ``has(name)``
|
|
to check before access when the domain might be absent.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._calendar: CalendarService | None = None
|
|
self._market: MarketService | None = None
|
|
self._cart: CartService | None = None
|
|
self._federation: FederationService | None = None
|
|
self._likes: LikesService | None = None
|
|
self._extra: dict[str, Any] = {}
|
|
|
|
# -- calendar -------------------------------------------------------------
|
|
@property
|
|
def calendar(self) -> CalendarService:
|
|
if self._calendar is None:
|
|
raise RuntimeError("CalendarService not registered")
|
|
return self._calendar
|
|
|
|
@calendar.setter
|
|
def calendar(self, impl: CalendarService) -> None:
|
|
self._calendar = impl
|
|
|
|
# -- market ---------------------------------------------------------------
|
|
@property
|
|
def market(self) -> MarketService:
|
|
if self._market is None:
|
|
raise RuntimeError("MarketService not registered")
|
|
return self._market
|
|
|
|
@market.setter
|
|
def market(self, impl: MarketService) -> None:
|
|
self._market = impl
|
|
|
|
# -- cart -----------------------------------------------------------------
|
|
@property
|
|
def cart(self) -> CartService:
|
|
if self._cart is None:
|
|
raise RuntimeError("CartService not registered")
|
|
return self._cart
|
|
|
|
@cart.setter
|
|
def cart(self, impl: CartService) -> None:
|
|
self._cart = impl
|
|
|
|
# -- likes ----------------------------------------------------------------
|
|
@property
|
|
def likes(self) -> LikesService:
|
|
if self._likes is None:
|
|
raise RuntimeError("LikesService not registered")
|
|
return self._likes
|
|
|
|
@likes.setter
|
|
def likes(self, impl: LikesService) -> None:
|
|
self._likes = impl
|
|
|
|
# -- federation -----------------------------------------------------------
|
|
@property
|
|
def federation(self) -> FederationService:
|
|
if self._federation is None:
|
|
raise RuntimeError("FederationService not registered")
|
|
return self._federation
|
|
|
|
@federation.setter
|
|
def federation(self, impl: FederationService) -> None:
|
|
self._federation = impl
|
|
|
|
# -- generic registration --------------------------------------------------
|
|
def register(self, name: str, impl: Any) -> None:
|
|
"""Register a service by name (for services without typed properties)."""
|
|
self._extra[name] = impl
|
|
|
|
def __getattr__(self, name: str) -> Any:
|
|
# Fallback to _extra dict for dynamically registered services
|
|
try:
|
|
extra = object.__getattribute__(self, "_extra")
|
|
if name in extra:
|
|
return extra[name]
|
|
except AttributeError:
|
|
pass
|
|
raise AttributeError(f"No service registered as: {name}")
|
|
|
|
# -- introspection --------------------------------------------------------
|
|
def has(self, name: str) -> bool:
|
|
"""Check whether a domain service is registered."""
|
|
if getattr(self, f"_{name}", None) is not None:
|
|
return True
|
|
return name in self._extra
|
|
|
|
|
|
# Module-level singleton — import this everywhere.
|
|
services = _ServiceRegistry()
|