Services

Learn how to create, register, and manage services in Singularity — including auto-discovery, HTTP routing, WebSocket endpoints, RPC exposure, and CLI scaffolding.

Services are the core building blocks of a Singularity application. Each service is a self-contained module that encapsulates business logic, exposes HTTP endpoints, and optionally participates in the RPC mesh. Singularity automatically discovers, registers, and wires services at startup with zero manual configuration.

How Auto-Discovery Works

When the application starts, the Manager recursively scans services/ for service.py files. Every directory that contains a service.py is treated as a service. Directories starting with __ (like __base and __pycache__) and entries listed in the .services_disabled file are skipped.

The module path is constructed from the directory structure. A service at services/v1/payments/service.py becomes module services.v1.payments.service and gets the API prefix /api/v1/payments. The class name must end with Service (e.g., PaymentsService).

Service Lifecycle

Services move through a well-defined set of states from creation to deletion:

State	Description
Created	The service directory and `service.py` file exist on disk but the app has not been restarted yet.
Active	The Manager discovered the service and registered its routes. It is handling traffic.
Disabled	The service path is listed in `.services_disabled`. It is skipped during discovery but files remain on disk.
Deleted	The service directory has been removed from disk entirely (hard delete).

Creating a Service

Via the CLI (Recommended)

The singularity create command scaffolds a new service with the correct directory structure, __init__.py files, and a populated service.py. It supports four base templates and four composable mixins.

# Basic CRUD service
singularity generate service v1/users

# With a description
singularity generate service v1/users -d "User management and profiles"

# Minimal read-only service
singularity generate service v1/health --template minimal

# Webhook receiver
singularity generate service v1/stripe --template webhook

# Service with RPC + database access
singularity generate service v1/payments --template crud --rpc --db

# Service with WebSocket + auth
singularity generate service v1/chat --template minimal --websocket --auth

Manually

Create the directory and file by hand:

Create the directory: services/v1/payments/
Add an __init__.py file in each new directory.
Create service.py with a class whose name ends in Service.
Restart the app. The Manager will discover and register it automatically.

Templates

Templates define the starting shape of a service. Each template pre-configures which HTTP methods are scaffolded and whether Acquire (the dependency injection container) is injected.

Full CRUD with Acquire injection. Best for data-driven services.

1"""User management and profiles"""23from singularity.core.acquire import Acquire456class UsersService:7"""User management and profiles"""89def __init__(self, acquire: Acquire):10  self.acquire = acquire1112async def get(self):13  """Get resource."""14  return {"status": "ok"}1516async def post(self, data: dict):17  """Create resource."""18  return {"status": "created", "data": data}1920async def put(self, data: dict):21  """Update resource."""22  return {"status": "updated", "data": data}2324async def delete(self):25  """Delete resource."""26  return {"status": "deleted"}

Mixins

Mixins are composable flags that layer additional functionality on top of any base template. You can combine multiple mixins in a single command.

Mixin	Flag	What it adds
RPC	`--rpc`	Sets `rpc_exposed = True` on the class, making it callable via `rpc("caller").service_name.method()`.
WebSocket	`--websocket` or `--ws`	Adds `http_exposed = ["ws=connect"]`, imports `WebSocket`, and generates a `ws_connect` handler method.
Auth	`--auth`	Imports `JWTBearer`, replaces the default `get` method with an authenticated version using `Depends(JWTBearer())`.
Database	`--db`	Adds the `Acquire` import (if not already present) so the service has access to `self.acquire.db_session`.

# Combine multiple mixins
singularity generate service v1/orders --template crud --rpc --db --auth

The generated code when combining --rpc, --auth, and --db:

"""Order processing service"""

from fastapi import Depends
from singularity.security import JWTBearer
from singularity.core.acquire import Acquire


class OrdersService:
  """Order processing service"""

  rpc_exposed = True

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

  async def get(self, payload: dict = Depends(JWTBearer())):
    """Get resource (authenticated)."""
    return {"status": "ok", "user": payload.get("id")}

  async def post(self, data: dict):
    """Create resource."""
    return {"status": "created", "data": data}

  async def put(self, data: dict):
    """Update resource."""
    return {"status": "updated", "data": data}

  async def delete(self):
    """Delete resource."""
    return {"status": "deleted"}

HTTP Method Routing

Core Methods

The Manager automatically registers any of the standard HTTP methods found on the service class. These are mapped to the service's base path:

Method on class	HTTP verb	Endpoint
`get(self)`	`GET`	`/api/v1/users`
`post(self, data)`	`POST`	`/api/v1/users`
`put(self, data)`	`PUT`	`/api/v1/users`
`delete(self)`	`DELETE`	`/api/v1/users`

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

  async def get(self):
    """List all users."""
    async with self.db() as session:
      result = await session.execute(select(User))
      return {"users": [u.to_dict() for u in result.scalars().all()]}

  async def post(self, data: dict):
    """Create a new user."""
    async with self.db() as session:
      user = User(**data)
      session.add(user)
      await session.commit()
      return {"status": "created", "id": user.id}

  async def put(self, data: dict):
    """Update an existing user."""
    return {"status": "updated", "data": data}

  async def delete(self):
    """Delete a user."""
    return {"status": "deleted"}

Custom Routes via `http_exposed`

For endpoints beyond basic CRUD, define an http_exposed list on the class. Each entry uses the format "method=sub_path". The Manager creates a route at /api/{service_path}/{sub_path} and looks for a handler method named {method}_{sub_path}.

class PaymentsService:
  http_exposed = ["get=status", "post=refund", "get=history"]

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

  async def get(self):
    """List payments — GET /api/v1/payments"""
    return {"payments": []}

  async def get_status(self, payment_id: str):
    """Check payment status — GET /api/v1/payments/status?payment_id=xxx"""
    return {"payment_id": payment_id, "status": "completed"}

  async def post_refund(self, payment_id: str):
    """Initiate a refund — POST /api/v1/payments/refund?payment_id=xxx"""
    return {"payment_id": payment_id, "refunded": True}

  async def get_history(self, user_id: str):
    """Get payment history — GET /api/v1/payments/history?user_id=xxx"""
    return {"user_id": user_id, "transactions": []}

WebSocket Endpoints

WebSocket routes are declared using the ws= prefix in http_exposed. The Manager registers them as WebSocket routes instead of standard HTTP routes.

from fastapi import WebSocket

class ChatService:
  http_exposed = ["ws=connect", "ws=stream"]

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

  async def get(self):
    """List active chat rooms — GET /api/v1/chat"""
    return {"rooms": []}

  async def ws_connect(self, websocket: WebSocket):
    """WebSocket — ws://host/api/v1/chat/connect"""
    await websocket.accept()
    try:
      while True:
        data = await websocket.receive_text()
        await self.ws_manager.broadcast({"message": data})
    except Exception:
      pass

  async def ws_stream(self, websocket: WebSocket):
    """WebSocket — ws://host/api/v1/chat/stream"""
    await websocket.accept()
    try:
      while True:
        await websocket.send_json({"type": "heartbeat"})
        await asyncio.sleep(5)
    except Exception:
      pass

The Manager wraps WebSocket handlers with error handling that logs exceptions and closes the connection with code 4000 if the handler raises.

RPC-Exposed Services

Any service can participate in the inter-service RPC mesh by setting rpc_exposed = True. This registers the service in the RPCRegistry, making its methods callable from other services via rpc("caller").target.method().

from singularity.rpc import rpc

class PaymentsService:
  """Payment processing service."""
  rpc_exposed = True
  blacklist = ["public", "analytics"]  # These callers are denied access

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

  async def charge(self, user_id: str, amount: float):
    """Charge a user."""
    return {"charged": amount, "user": user_id}

  async def refund(self, payment_id: str):
    """Refund a payment."""
    return {"refunded": True, "payment": payment_id}


class OrdersService:
  """Order management — calls PaymentsService via RPC."""

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

  async def post(self, data: dict):
    """Create an order and charge the user."""
    payment = await rpc("orders").payments.charge(
      user_id=data["user_id"],
      amount=data["total"],
    )
    return {"order": "confirmed", "payment": payment}

The blacklist is enforced at call time. If "analytics" is in the blacklist, then rpc("analytics").payments.charge(...) raises RPCAccessDenied. See the RPC guide for full details on blacklisting, remote calls, and the heartbeat system.

Hooks

Services support per-method before/after/error hooks for composable interception -- authentication, validation, logging, error recovery, and more. Hooks are declared as a class attribute and run around service method calls with zero overhead for services that don't use them.

from singularity.core.hooks import HookContext

async def authenticate(ctx: HookContext):
    if not ctx.kwargs.get("token"):
        from fastapi import HTTPException
        raise HTTPException(status_code=401)

class UsersService:
    hooks = {
        "before": {"all": [authenticate]},
    }

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

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

Services can also define async setup() and teardown() lifecycle methods, and declare service_events for automatic WebSocket broadcasts on CRUD operations.

See the full Hooks & Lifecycle guide for execution flow diagrams, all hook patterns, service events, and complete examples.

The Acquire Container

Acquire is the dependency injection container that the Manager passes into services whose __init__ accepts an acquire parameter. It provides access to all shared resources:

Property	Type	Description
`acquire.db_session`	`async_session`	SQLAlchemy async session factory
`acquire.settings`	`Settings`	Pydantic application settings
`acquire.logger`	`Logger`	Loguru logger instance
`acquire.cache`	`Cache`	Cache utility
`acquire.deps_cache`	`deps_cache`	Dependency-level cache
`acquire.ws_manager`	`WebSocketManager`	WebSocket connection manager
`acquire.tasks`	`TaskRunner`	Background task runner
`acquire.schemas`	`dict`	Auto-discovered Pydantic schemas
`acquire.services`	`dict`	All discovered service instances
`acquire.utils`	`module`	Utility helpers and auth tools

class NotificationService:
  def __init__(self, acquire: Acquire):
    self.acquire = acquire
    self.db = acquire.db_session
    self.ws = acquire.ws_manager
    self.tasks = acquire.tasks

  async def post(self, data: dict):
    """Send a notification."""
    user_id = data["user_id"]
    message = data["message"]

    # Store in database
    async with self.db() as session:
      session.add(Notification(user_id=user_id, text=message))
      await session.commit()

    # Push in real-time via WebSocket
    await self.ws.broadcast({"type": "notification", "message": message})

    # Queue a follow-up background task
    self.tasks.submit("send_push_notification", user_id=user_id, message=message)

    return {"status": "sent"}

Managing Services

Listing Services

# List all active services
services list

# Detailed view with methods, RPC status, and endpoints
services list --detailed

# Include disabled services
services list --all

Getting Service Info

# Full info for a specific service
services info payments

This displays the class name, module path, API endpoint, HTTP methods, RPC status, blacklist, WebSocket routes, and dependency information.

Disabling and Enabling

# Soft-disable (adds to .services_disabled, files stay on disk)
services delete v1/analytics --soft

# Re-enable a disabled service
services enable v1/analytics

Deleting Services

# Hard delete (removes all files permanently)
services delete old_service --hard

# Hard delete with confirmation bypass
services delete old_service --hard --force

Hard deletes are irreversible. The service directory and all its contents are permanently removed from disk. Use soft-disable if you might need the service again.

Validating Before Creating

The validate command performs a dry run of the generator, showing you exactly what would be created without writing any files:

services validate v1/analytics --template crud --rpc

This outputs the generated file content to the terminal so you can review it before committing.

Dependency Graph

The singularity services graph command visualizes how services depend on each other through RPC calls and remote microservice connections:

# Print ASCII dependency graph
services graph

# Output DOT format for Graphviz rendering
services graph --dot | dot -Tpng -o deps.png

# Machine-readable JSON
services graph --json

Directory Structure Convention

Services follow a convention-based directory layout. The path determines the API endpoint:

services/
  __base/           # Framework internals (skipped by discovery)
  ws/               # Core WebSocket service (skipped unless --include-core)
  v1/
    users/
      __init__.py
      service.py    # -> /api/v1/users
      schema.py     # Pydantic schemas (auto-discovered by Acquire)
    payments/
      __init__.py
      service.py    # -> /api/v1/payments
    orders/
      __init__.py
      service.py    # -> /api/v1/orders
  v2/
    users/
      __init__.py
      service.py    # -> /api/v2/users
  .services_disabled  # One service path per line

The .services_disabled file contains one service path per line (e.g., v1/analytics). Comments starting with # are ignored. This file is managed by the CLI but can be edited by hand.

Complete Example: E-Commerce Service

Here is a full-featured service that combines CRUD operations, custom routes, WebSocket events, RPC calls, and database access:

"""Order management service with payment integration."""

from fastapi import WebSocket, Depends
from singularity.security import JWTBearer
from singularity.core.acquire import Acquire
from singularity.rpc import rpc


class OrdersService:
  """Order management with real-time updates and payment processing."""

  rpc_exposed = True
  blacklist = ["public"]
  http_exposed = ["get=status", "post=cancel", "ws=updates"]

  def __init__(self, acquire: Acquire):
    self.acquire = acquire
    self.db = acquire.db_session
    self.ws = acquire.ws_manager
    self.settings = acquire.settings

  async def get(self, payload: dict = Depends(JWTBearer())):
    """List orders for the authenticated user."""
    user_id = payload.get("id")
    async with self.db() as session:
      result = await session.execute(
        select(Order).where(Order.user_id == user_id)
      )
      return {"orders": [o.to_dict() for o in result.scalars().all()]}

  async def post(self, data: dict):
    """Create a new order and charge via RPC."""
    # Call PaymentsService via RPC
    payment = await rpc("orders").payments.charge(
      user_id=data["user_id"],
      amount=data["total"],
    )

    # Persist the order
    async with self.db() as session:
      order = Order(
        user_id=data["user_id"],
        total=data["total"],
        payment_ref=payment.get("transaction_id"),
      )
      session.add(order)
      await session.commit()

    # Notify connected clients
    await self.ws.broadcast({
      "type": "order_created",
      "order_id": order.id,
    })

    return {"order_id": order.id, "payment": payment}

  async def get_status(self, order_id: str):
    """Check order status — GET /api/v1/orders/status"""
    return {"order_id": order_id, "status": "processing"}

  async def post_cancel(self, order_id: str):
    """Cancel an order — POST /api/v1/orders/cancel"""
    refund = await rpc("orders").payments.refund(payment_id=order_id)
    return {"order_id": order_id, "cancelled": True, "refund": refund}

  async def ws_updates(self, websocket: WebSocket):
    """Real-time order updates — ws://host/api/v1/orders/updates"""
    await websocket.accept()
    try:
      while True:
        data = await websocket.receive_text()
        await websocket.send_json({"type": "ack", "data": data})
    except Exception:
      pass

Scaffold an equivalent starting point with:

singularity generate service v1/orders -d "Order management" --template crud --rpc --websocket --auth --db

What Happens at Startup

When you run uv run python app.py --dev, the following sequence occurs:

Manager(app) is created with a reference to the FastAPI application and an Acquire instance.
manager.register_middlewares() scans singularity/middleware/ and registers any middleware classes found.
manager.register_services() recursively discovers all services, registers HTTP routes, and collects RPC-exposed services.
_init_rpc() registers all collected RPC services in the RPCRegistry and generates the RPC spec.
FastAPI's lifespan fires manager.startup():
- Calls setup() on services that define it (in registration order).
- Publishes the RPC spec to Redis with a 60-second TTL.
- Discovers remote services from Redis.
- Starts the heartbeat loop (every 30 seconds) unless --no-heartbeat is passed.
The server begins accepting requests.
On shutdown, manager.shutdown():
- Calls teardown() on services that define it (in reverse registration order).
- Cancels the heartbeat and removes the spec from Redis.

On this page