Hooks & Lifecycle

Add per-service before/after/error hooks, lifecycle methods, and auto-emitting service events to Singularity services.

Hooks are composable middleware that run around individual service methods. Unlike ASGI middleware (which wraps every request globally), hooks are declared per-service and can target specific HTTP methods. They are inspired by FeatherJS hooks.

Singularity hooks provide three phases:

Phase	When it runs	Use case
before	Before the service method executes	Authentication, input validation, rate limiting
after	After the service method returns	Response transformation, logging, caching
error	When the service method raises an exception	Error recovery, fallback responses, error logging

Declaring Hooks

Hooks are declared as a hooks class attribute on your service. Each phase maps method names (or "all") to a list of async hook functions:

from singularity.core.hooks import HookContext

async def authenticate(ctx: HookContext):
    """Reject unauthenticated requests."""
    token = ctx.kwargs.get("token")
    if not token:
        from fastapi import HTTPException
        raise HTTPException(status_code=401, detail="Not authenticated")

async def log_response(ctx: HookContext):
    """Log every response."""
    print(f"{ctx.service_name}.{ctx.method} -> {ctx.result}")

async def require_admin(ctx: HookContext):
    """Only allow admin users to delete."""
    user = ctx.kwargs.get("user", {})
    if user.get("role") != "admin":
        from fastapi import HTTPException
        raise HTTPException(status_code=403, detail="Admin required")

class UsersService:
    hooks = {
        "before": {
            "all": [authenticate],        # runs before every method
            "delete": [require_admin],     # runs before delete only
        },
        "after": {
            "all": [log_response],         # runs after every method
        },
        "error": {
            "all": [handle_error],         # runs on any exception
        },
    }

    def __init__(self, acquire):
        self.acquire = acquire

    async def get(self):
        return {"users": []}

    async def post(self, data: dict):
        return {"created": data}

    async def delete(self, user_id: str):
        return {"deleted": user_id}

The "all" key targets every method. Method-specific hooks run after the "all" hooks for that phase.

HookContext

Every hook function receives a HookContext dataclass as its sole argument:

from singularity.core.hooks import HookContext

Field	Type	Description
`service`	`Any`	The service instance
`service_name`	`str`	Service name (e.g. `"users"`)
`method`	`str`	HTTP method name (e.g. `"get"`, `"post"`)
`args`	`tuple`	Positional args passed to the endpoint
`kwargs`	`dict[str, Any]`	Keyword args (mutable -- before hooks can modify)
`result`	`Any`	Method result; after hooks can modify
`error`	`Exception \| None`	Exception, if any. Set in error hooks
`acquire`	`Any`	The Acquire instance (if the service has one)

Execution Flow

Key behaviors:

Before hooks run in order: all first, then method-specific
Short-circuit: If a before hook sets ctx.result, the service method is skipped (after hooks still run)
After hooks can read and modify ctx.result
Error hooks receive the exception in ctx.error. Setting ctx.result swallows the error and returns that value instead
Zero overhead: Services without hooks have no wrapping at all -- the Manager returns the original endpoint unchanged

Common Patterns

Authentication

async def authenticate(ctx: HookContext):
    """Check for a valid JWT token."""
    token = ctx.kwargs.get("authorization")
    if not token:
        from fastapi import HTTPException
        raise HTTPException(status_code=401)
    # Decode and attach user info
    ctx.kwargs["current_user"] = decode_jwt(token)

Input Transformation

async def normalize_email(ctx: HookContext):
    """Lowercase email before processing."""
    data = ctx.kwargs.get("data", {})
    if "email" in data:
        data["email"] = data["email"].lower()

Response Wrapping

async def wrap_response(ctx: HookContext):
    """Wrap all responses in a standard envelope."""
    ctx.result = {
        "success": True,
        "data": ctx.result,
        "service": ctx.service_name,
    }

Short-Circuiting (Cache Hit)

async def check_cache(ctx: HookContext):
    """Return cached result if available."""
    cache_key = f"{ctx.service_name}:{ctx.method}"
    cached = await get_from_cache(cache_key)
    if cached is not None:
        ctx.result = cached  # Skips the service method entirely

Error Recovery

async def fallback_on_error(ctx: HookContext):
    """Return a fallback response instead of crashing."""
    ctx.result = {
        "error": str(ctx.error),
        "fallback": True,
    }

Service Lifecycle

Services can define async setup() and teardown() methods that run during application startup and shutdown:

class AnalyticsService:
    def __init__(self, acquire):
        self.acquire = acquire
        self.client = None

    async def setup(self):
        """Called once on application startup."""
        self.client = await create_analytics_client()

    async def teardown(self):
        """Called once on application shutdown."""
        if self.client:
            await self.client.close()

    async def post(self, data: dict):
        await self.client.track(data)
        return {"tracked": True}

The Manager automatically detects services with setup() or teardown() methods:

setup() is called during manager.startup(), in registration order
teardown() is called during manager.shutdown(), in reverse registration order
Errors in one service's lifecycle do not prevent other services from running

Lifecycle methods are ideal for initializing connections, warming caches, or registering with external systems that require async initialization.

Service Events

Services can automatically broadcast events over WebSocket when CRUD operations succeed. Declare a service_events list with the event types you want to emit:

class OrdersService:
    service_events = ["created", "updated", "removed"]

    def __init__(self, acquire):
        self.acquire = acquire

    async def post(self, data: dict):
        # After this returns, "orders.created" is broadcast automatically
        return {"order_id": "123", "data": data}

    async def put(self, data: dict):
        # After this returns, "orders.updated" is broadcast automatically
        return {"updated": True}

    async def delete(self):
        # After this returns, "orders.removed" is broadcast automatically
        return {"deleted": True}

The event mapping is:

HTTP Method	Event Name	Broadcast Payload
`post`	`{service}.created`	The method's return value
`put` / `patch`	`{service}.updated`	The method's return value
`delete`	`{service}.removed`	The method's return value

Events are broadcast via acquire.ws_manager.broadcast(). If the service has no acquire or no ws_manager, event emission is silently skipped.

Service events are implemented as internal after-hooks appended by the Manager. They run after any user-defined after hooks and have zero overhead for services that don't declare service_events.

Combining Hooks, Lifecycle, and Events

Here is a complete example combining all three features:

from singularity.core.hooks import HookContext

async def validate_order(ctx: HookContext):
    """Ensure order data has required fields."""
    data = ctx.kwargs.get("data", {})
    if "items" not in data or not data["items"]:
        from fastapi import HTTPException
        raise HTTPException(status_code=400, detail="Order must have items")

async def enrich_response(ctx: HookContext):
    """Add timestamp to all responses."""
    from datetime import datetime
    if isinstance(ctx.result, dict):
        ctx.result["timestamp"] = datetime.utcnow().isoformat()

class OrdersService:
    """Order management with hooks, lifecycle, and events."""

    rpc_exposed = True
    service_events = ["created", "updated"]

    hooks = {
        "before": {
            "post": [validate_order],
        },
        "after": {
            "all": [enrich_response],
        },
    }

    def __init__(self, acquire):
        self.acquire = acquire
        self.db = acquire.db_session

    async def setup(self):
        """Warm the order cache on startup."""
        # Pre-load frequently accessed data
        pass

    async def teardown(self):
        """Flush pending analytics on shutdown."""
        pass

    async def get(self):
        return {"orders": []}

    async def post(self, data: dict):
        # validate_order runs first (before hook)
        # then this method runs
        # then enrich_response adds timestamp (after hook)
        # then "orders.created" is broadcast (service event)
        return {"order_id": "new", "items": data["items"]}

    async def put(self, data: dict):
        return {"updated": True}

On this page