Service API

Complete API reference for the service layer — Manager, Acquire, ServiceDiscovery, ServiceGenerator, BaseWebhook, and WebSocketManager.

This page provides structured API tables for every class in the Singularity service layer. All signatures are taken directly from the source code in singularity/core/ and singularity/common/websocket.py.

Manager

services.__base.manager.Manager

The Manager is the central orchestrator. It scans for services and middlewares, registers routers with FastAPI, initializes RPC, and manages the application lifespan (startup/shutdown).

Constructor

Parameter	Type	Default	Description
`app`	`FastAPI`	(required)	The FastAPI application instance
`prefix`	`str`	`"/api"`	Base URL prefix for all service routers

Attributes

Attribute	Type	Description
`app`	`FastAPI`	The FastAPI application instance
`prefix`	`str`	Base URL prefix for all services
`acquire`	`Acquire`	The shared dependency injection container
`services_dir`	`str`	Absolute path to the `services/` directory
`mws_dir`	`str`	Absolute path to the `singularity/middleware/` directory
`ws_routes`	`Dict[str, Type]`	Registered WebSocket route endpoints

Methods

Method	Parameters	Return Type	Description
`register_services()`	(none)	`None`	Recursively scan `services_dir`, import each `service.py`, inject `Acquire`, register routers and RPC entries.
`register_middlewares()`	(none)	`None`	Scan `mws_dir` for middleware modules, instantiate each `Middleware` class, and add to the FastAPI app.
`startup()`	(none)	`None`	Async. Publishes the RPC spec to Redis, discovers remote services, and starts the heartbeat loop (30s interval, 60s TTL).
`shutdown()`	(none)	`None`	Async. Cancels the heartbeat task and removes RPC specs from Redis.
`register_ws_routes(router, service_instance, service_name)`	`router: APIRouter`, `service_instance: Any`, `service_name: str`	`None`	Extract `ws=` entries from `http_exposed` and register WebSocket endpoints on the router.

register_services() also calls _init_rpc() internally, which registers all rpc_exposed services with the RPCRegistry. You do not need to call _init_rpc() yourself.

Acquire

services.__base.acquire.Acquire

The Acquire class is the dependency injection container. A single instance is created by the Manager and passed to every service whose __init__ accepts an acquire parameter.

Constructor

Takes no parameters. All resources are initialized internally.

Available Resources

Resource	Type	Description
`db_session`	`async_sessionmaker`	Async database session factory (SQLAlchemy `async_sessionmaker`)
`schemas`	`dict`	Auto-discovered schema classes (loaded from `schema.py` files)
`services`	`dict`	Auto-discovered service classes (loaded from `service.py` files)
`settings`	`Settings`	Pydantic settings instance (loaded from `.env`)
`utils`	`module`	The utils module (auth helpers, module loader, etc.)
`logger`	`loguru.Logger`	Pre-configured Loguru logger instance
`cache`	`Cache`	In-memory cache instance
`deps_cache`	`deps_cache`	Dependency-level cache instance
`ws_manager`	`WebSocketManager`	WebSocket connection manager
`tasks`	`TaskRunner`	Background task runner (auto-discovers tasks on init)

Services that do not need dependency injection can omit acquire from their __init__ signature entirely. The Manager will detect this and instantiate the service without it.

ServiceDiscovery

services.__base.discovery.ServiceDiscovery

Utility for discovering and inspecting services by scanning the services/ directory. Used by the CLI, not at runtime.

Constructor

Takes no parameters. Automatically resolves services_dir from its own file location.

Methods

Method	Parameters	Return Type	Description
`discover_all(include_core)`	`include_core: bool = False`	`list[ServiceInfo]`	Scan the services directory recursively and return metadata for all discovered services. Core services (e.g. `ws`) are excluded unless `include_core=True`.
`get_service_info(name)`	`name: str`	`Optional[ServiceInfo]`	Get info for a specific service by name or path (e.g. `"payments"` or `"v1/payments"`). Includes core services in lookup.
`build_dependency_graph(include_core)`	`include_core: bool = False`	`dict`	Build a dependency graph with keys: `nodes`, `edges` (RPC), `remote_edges` (Microservice), `blocked` (blacklist entries).
`get_disabled_services()`	(none)	`set[str]`	Read the set of disabled service paths from `.services_disabled`.
`disable_service(name)`	`name: str`	`None`	Add a service to the disabled list.
`enable_service(name)`	`name: str`	`None`	Remove a service from the disabled list.
`is_disabled(name)`	`name: str`	`bool`	Check if a service is currently disabled.
`service_exists(name)`	`name: str`	`bool`	Check if a service directory with `service.py` exists on disk.

Dependency Graph Return Structure

The build_dependency_graph() method returns a dictionary with the following shape:

{
    "nodes": [
        {"name": "payments", "path": "v1/payments", "rpc_exposed": True, "is_webhook": False}
    ],
    "edges": [
        {"from": "orders", "to": "payments", "type": "rpc"}
    ],
    "remote_edges": [
        {"from": "orders", "to": "billing", "host_url": "http://...", "service_name": "billing", "type": "remote"}
    ],
    "blocked": [
        {"from": "payments", "blocked": "public", "type": "blacklist"}
    ]
}

ServiceInfo

services.__base.discovery.ServiceInfo

A dataclass holding metadata about a discovered service. Returned by ServiceDiscovery.discover_all() and ServiceDiscovery.get_service_info().

Fields

Field	Type	Default	Description
`name`	`str`	(required)	The service name (last path segment, e.g. `"payments"`)
`path`	`Path`	(required)	Absolute filesystem path to `service.py`
`path_segments`	`list[str]`	(required)	Directory segments (e.g. `["v1", "payments"]`)
`module_path`	`str`	(required)	Python import path (e.g. `"services.v1.payments.service"`)
`api_endpoint`	`str`	(required)	API URL (e.g. `"/api/v1/payments"`)
`class_name`	`str`	(required)	The service class name (e.g. `"PaymentsService"`)
`methods`	`list[str]`	`[]`	HTTP methods (e.g. `["GET", "POST"]`)
`http_exposed`	`list[str]`	`[]`	Custom route declarations (e.g. `["get=status", "ws=events"]`)
`rpc_exposed`	`bool`	`False`	Whether the service is exposed for RPC
`blacklist`	`list[str]`	`[]`	Services blocked from calling this service via RPC
`docstring`	`Optional[str]`	`None`	First line of the service class docstring
`is_disabled`	`bool`	`False`	Whether the service is in the disabled list
`is_core`	`bool`	`False`	Whether the service is a framework core service
`is_webhook`	`bool`	`False`	Whether the service inherits from `BaseWebhook` or defines `events`
`webhook_events`	`list[str]`	`[]`	Event type keys from the `events` dict
`rpc_dependencies`	`list[str]`	`[]`	Services this service depends on via RPC descriptors
`remote_dependencies`	`list[dict[str, Any]]`	`[]`	Remote microservice dependencies (`attr`, `host_url`, `service_name`)

ServiceGenerator

services.__base.generator.ServiceGenerator

Generator for creating new services with composable templates and mixins. Used by the services create CLI command.

Constructor

Takes no parameters. Resolves services_dir from its own file location.

Methods

Method	Parameters	Return Type	Description
`validate_service_name(name)`	`name: str`	`tuple[bool, Optional[str]]`	Validate a service name. Returns `(True, None)` on success or `(False, error_message)` on failure. Checks snake_case, reserved names, Python keywords, and existence.
`generate_class_name(service_name)`	`service_name: str`	`str`	Generate a PascalCase class name from the last path segment (e.g. `"v1/user_profiles"` becomes `"UserProfilesService"`).
`generate_template(...)`	See table below	`str`	Generate the full Python source code for a service file.
`create_service(...)`	See table below	`Path`	Create the service directory, write `service.py`, and return the file path.
`delete_service(name, hard)`	`name: str`, `hard: bool`	`Path`	Delete (hard) or soft-disable a service. Returns the service directory path.

`generate_template()` / `create_service()` Parameters

Parameter	Type	Default	Description
`name`	`str`	(required)	Service path (e.g. `"v1/users"`)
`description`	`str`	(required)	Service description for the docstring
`base`	`str`	`"crud"`	Base template: `"crud"`, `"minimal"`, `"empty"`, or `"webhook"`
`methods`	`Optional[list[str]]`	`None`	Override template methods (e.g. `["get", "post"]`). `None` uses template defaults.
`rpc`	`bool`	`False`	Add `rpc_exposed = True` class attribute
`websocket`	`bool`	`False`	Add WebSocket route (`ws=connect`) and `ws_connect` method
`auth`	`bool`	`False`	Add JWT authentication (replaces `get` with authenticated version)
`db`	`bool`	`False`	Add database session imports
`overwrite`	`bool`	`False`	Allow overwriting existing service (`create_service` only)

Base Templates

	crud	minimal	empty	webhook
Methods	get, post, put, delete	get	(none)	(none)
Acquire	Yes	No	No	Yes
Use case	Full CRUD service	Read-only endpoint	Custom service shell	Webhook receiver

BaseWebhook

services.__base.webhook.BaseWebhook

Base class for webhook receiver services. Handles raw body extraction, signature verification, event type extraction, and dispatch to user-defined handler methods.

Constructor

Parameter	Type	Default	Description
`acquire`	`Acquire`	(required)	The dependency injection container

Class Attributes

Attribute	Type	Default	Description
`rpc_exposed`	`bool`	`False`	Whether this webhook is exposed via RPC (typically left `False`)
`events`	`dict[str, str]`	`{}`	Mapping of event type strings to handler method names

Methods

Method	Parameters	Return Type	Description
`post(request)`	`request: Request`	`JSONResponse`	Async. Webhook entry point registered by the Manager as a POST endpoint. Executes: raw body extraction, `verify()`, JSON parse, `get_event_type()`, handler dispatch.
`verify(headers, raw_body)`	`headers: dict`, `raw_body: bytes`	`bool`	Verify the webhook signature. Override for platform-specific verification (HMAC-SHA256 for Stripe, SHA-1 for GitHub, etc.). Returns `True` by default.
`get_event_type(headers, payload)`	`headers: dict`, `payload: dict`	`str`	Extract the event type string from the webhook. Override for platform-specific extraction. Returns `payload.get("type", "unknown")` by default.
`on_unhandled_event(event_type, payload, headers)`	`event_type: str`, `payload: dict`, `headers: dict`	`JSONResponse`	Async. Called when an event type has no mapping in `self.events`. Default: log and return `200 OK` with `{"status": "ignored"}`.

Response Status Codes

Scenario	Status Code	Body
`verify()` returns `False`	`401`	`{"error": "Verification failed"}`
Invalid JSON body	`400`	`{"error": "Invalid JSON payload"}`
Unhandled event type	`200`	`{"status": "ignored", "event": "<type>"}`
Handler method missing from class	`500`	`{"error": "Handler misconfigured"}`
Handler raises exception	`500`	`{"error": "Internal handler error"}`
Handler succeeds	`200`	Handler return value

BaseWebhook does not provide built-in idempotency. You are responsible for deduplicating events in your handler methods (e.g. by tracking processed event IDs in your database).

Field	Type	Description
`websocket`	`WebSocket`	The FastAPI WebSocket instance
`user_id`	`UUID`	User identifier
`org_id`	`UUID`	Organization identifier
`channel_id`	`str`	Computed as `"{user_id}_{org_id}"`
`permissions`	`list`	Permission strings for RBAC filtering

`WebSocketMessageType` (Enum)

Member	Value	Description
`CONNECT`	`"connect"`	Connection confirmation
`SUBSCRIBE`	`"subscribe"`	Channel subscription
`UNSUBSCRIBE`	`"unsubscribe"`	Channel unsubscription
`ERROR`	`"error"`	Error notification
`MESSAGE`	`"message"`	Data message

Methods

Method	Parameters	Return Type	Description
`connect(websocket, org_id, user_id, permissions)`	`websocket: WebSocket`, `org_id: UUID`, `user_id: UUID`, `permissions: list`	`None`	Async. Accept the WebSocket connection, store it, send connection confirmation, and enter the receive loop. Automatically calls `disconnect` on `WebSocketDisconnect`.
`disconnect(channel_id)`	`channel_id: str`	`None`	Async. Remove a connection from tracking and clean up organization mappings.
`broadcast(org_id, data, resource, required_action, exclude_channel)`	`org_id: UUID`, `data: dict`, `resource: str`, `required_action: str`, `exclude_channel: Optional[str]`	`None`	Async. Broadcast a message to all eligible clients in an organization. Only sends to connections whose permissions include `"{resource}_{required_action}"`.

Manager

Constructor

Attributes

Methods

Acquire

Constructor

Available Resources

ServiceDiscovery

Constructor

Methods

Dependency Graph Return Structure

ServiceInfo

Fields

ServiceGenerator

Constructor

Methods

`generate_template()` / `create_service()` Parameters

Base Templates

BaseWebhook

Constructor

Class Attributes

Methods

Response Status Codes

WebSocketManager

Constructor

Internal Types

`WebSocketConnection` (dataclass)

`WebSocketMessageType` (Enum)

Methods

On this page