Service API Reference

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

appFastAPIrequired

The FastAPI application instance

prefixstr= "/api"

Base URL prefix for all service routers

Attributes

appFastAPI

The FastAPI application instance

prefixstr

Base URL prefix for all services

acquireAcquire

The shared dependency injection container

services_dirstr

Absolute path to the services/ directory

mws_dirstr

Absolute path to the singularity/middleware/ directory

ws_routesDict[str, Type]

Registered WebSocket route endpoints

Methods

register_services()None

Recursively scan services_dir, import each service.py, inject Acquire, register routers and RPC entries.

register_middlewares()None

Scan mws_dir for middleware modules, instantiate each Middleware class, and add to the FastAPI app.

startup()None

Async. Publishes RPC spec to Redis, discovers remote services, and starts the heartbeat loop (30s interval, 60s TTL).

shutdown()None

Async. Cancels the heartbeat task and removes RPC specs from Redis.

register_ws_routes(router, service_instance, service_name)None

Extract ws= entries from http_exposed and register WebSocket endpoints on the router. Parameters: router (APIRouter), service_instance (Any), service_name (str).

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

db_sessionasync_sessionmaker

Async database session factory (SQLAlchemy async_sessionmaker)

schemasdict

Auto-discovered schema classes (loaded from schema.py files)

servicesdict

Auto-discovered service classes (loaded from service.py files)

settingsSettings

Pydantic settings instance (loaded from .env)

utilsmodule

The utils module (auth helpers, module loader, etc.)

loggerloguru.Logger

Pre-configured Loguru logger instance

cacheCache

In-memory cache instance

deps_cachedeps_cache

Dependency-level cache instance

ws_managerWebSocketManager

WebSocket connection manager

tasksTaskRunner

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

discover_all(include_core)list[ServiceInfo]= include_core: bool = False

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)Optional[ServiceInfo]= name: str

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)dict= include_core: bool = False

Build a dependency graph with keys: nodes, edges (RPC), remote_edges (Microservice), blocked (blacklist entries).

get_disabled_services()set[str]

Read the set of disabled service paths from .services_disabled.

disable_service(name)None= name: str

Add a service to the disabled list.

enable_service(name)None= name: str

Remove a service from the disabled list.

is_disabled(name)bool= name: str

Check if a service is currently disabled.

service_exists(name)bool= name: str

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

namestrrequired

The service name (last path segment, e.g. "payments")

pathPathrequired

Absolute filesystem path to service.py

path_segmentslist[str]required

Directory segments (e.g. ["v1", "payments"])

module_pathstrrequired

Python import path (e.g. "services.v1.payments.service")

api_endpointstrrequired

API URL (e.g. "/api/v1/payments")

class_namestrrequired

The service class name (e.g. "PaymentsService")

methodslist[str]= []

HTTP methods (e.g. ["GET", "POST"])

http_exposedlist[str]= []

Custom route declarations (e.g. ["get=status", "ws=events"])

rpc_exposedbool= False

Whether the service is exposed for RPC

blacklistlist[str]= []

Services blocked from calling this service via RPC

docstringOptional[str]= None

First line of the service class docstring

is_disabledbool= False

Whether the service is in the disabled list

is_corebool= False

Whether the service is a framework core service

is_webhookbool= False

Whether the service inherits from BaseWebhook or defines events

webhook_eventslist[str]= []

Event type keys from the events dict

rpc_dependencieslist[str]= []

Services this service depends on via RPC descriptors

remote_dependencieslist[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

validate_service_name(name)tuple[bool, Optional[str]]= name: 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)str= service_name: str

Generate a PascalCase class name from the last path segment (e.g. "v1/user_profiles" becomes "UserProfilesService").

generate_template(...)str

Generate the full Python source code for a service file. See parameters below.

create_service(...)Path

Create the service directory, write service.py, and return the file path. See parameters below.

delete_service(name, hard)Path

Delete (hard) or soft-disable a service. Returns the service directory path.

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

namestrrequired

Service path (e.g. "v1/users")

descriptionstrrequired

Service description for the docstring

basestr= "crud"

Base template: "crud", "minimal", "empty", or "webhook"

methodsOptional[list[str]]= None

Override template methods (e.g. ["get", "post"]). None uses template defaults.

rpcbool= False

Add rpc_exposed = True class attribute

websocketbool= False

Add WebSocket route (ws=connect) and ws_connect method

authbool= False

Add JWT authentication (replaces get with authenticated version)

dbbool= False

Add database session imports

overwritebool= 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

acquireAcquirerequired

The dependency injection container

Class Attributes

rpc_exposedbool= False

Whether this webhook is exposed via RPC (typically left False)

eventsdict[str, str]= {}

Mapping of event type strings to handler method names

Methods

post(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)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)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)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).

WebSocketManager

common.websocket.WebSocketManager

WebSocket connection manager with organization-level RBAC integration. Manages connection lifecycle and message broadcasting.

Constructor

Takes no parameters. Initializes empty connection and organization tracking dictionaries.

Internal Types

`WebSocketConnection` (dataclass)

websocketWebSocket

The FastAPI WebSocket instance

user_idUUID

User identifier

org_idUUID

Organization identifier

channel_idstr

Computed as "{user_id}_{org_id}"

permissionslist

Permission strings for RBAC filtering

`WebSocketMessageType` (Enum)

CONNECT"connect"

Connection confirmation

SUBSCRIBE"subscribe"

Channel subscription

UNSUBSCRIBE"unsubscribe"

Channel unsubscription

ERROR"error"

Error notification

MESSAGE"message"

Data message

Methods

connect(websocket, org_id, user_id, permissions)None

Async. Accept the WebSocket connection, store it, send connection confirmation, and enter the receive loop. Automatically calls disconnect on WebSocketDisconnect.

disconnect(channel_id)None

Async. Remove a connection from tracking and clean up organization mappings.

broadcast(org_id, data, resource, required_action, exclude_channel)None

Async. Broadcast a message to all eligible clients in an organization. Only sends to connections whose permissions include "{resource}_{required_action}".

Service API Reference

On this page