Dependency Injection

Deep dive into the Acquire DI container, how the Manager injects dependencies into services, auto-discovery mechanics, and service name injection for RPC.

Singularity uses a lightweight, purpose-built dependency injection system centered around the Acquire container. Rather than a full DI framework with decorators, scopes, and binding tables, Acquire is a single object that holds all shared resources and gets passed to services that need it.

The Acquire Container

Acquire is defined in singularity/core/acquire.py. When instantiated, it eagerly initializes every shared resource the application needs:

# singularity/core/acquire.py
class Acquire:
    def __init__(self):
        self.db_session = async_session          # AsyncSession factory
        self.schemas = self._register_schemas()  # Auto-discovered schema classes
        self.services = self._register_services() # Auto-discovered service refs
        self.settings: Settings = settings       # Pydantic settings singleton
        self.utils = utils                       # Utility module (auth helpers, etc.)
        self.logger = logger                     # Loguru logger instance
        self.cache = Cache()                     # In-memory TTL cache
        self.deps_cache = deps_cache             # Async cache (Redis or in-memory)
        self.ws_manager = WebSocketManager()     # WebSocket connection manager
        self.tasks = TaskRunner()                # Background task runner
        self.tasks.discover_tasks()              # Auto-discover @task definitions

Acquire is instantiated once by the Manager and shared across all services as a singleton. This ensures every service gets the same database session factory, the same settings object, and the same task runner.

Dependency Graph

The following diagram shows how Acquire connects to infrastructure resources and how the Manager distributes it to services:

Available Resources

Every service that receives acquire has access to the following resources:

Resource	Type	Description
`acquire.db_session`	`async_sessionmaker[AsyncSession]`	Factory for creating async database sessions. Call `async with acquire.db_session() as session:` to get a session.
`acquire.settings`	`Settings`	Pydantic settings object with all environment configuration (database URL, JWT secret, Redis URLs, etc.).
`acquire.tasks`	`TaskRunner`	Submit background tasks to the Celery worker pool.
`acquire.ws_manager`	`WebSocketManager`	Manage WebSocket connections, broadcast messages to connected clients.
`acquire.cache`	`Cache`	Simple in-memory TTL cache for synchronous access patterns.
`acquire.deps_cache`	`CacheInterface`	Async cache backed by Redis (if configured) or in-memory fallback.
`acquire.logger`	`loguru.Logger`	Structured logger configured per environment (dev/beta/prod).
`acquire.schemas`	`dict`	Auto-discovered schema classes loaded from `schema.py` files in service directories.
`acquire.services`	`dict`	References to all discovered service classes (for internal use).
`acquire.utils`	`module`	Utility module containing `ModuleLoader`, auth helpers, and other shared functions.

How Manager Injects Acquire

The injection mechanism is straightforward: the Manager inspects each service class's __init__ signature at registration time. If the signature includes an acquire parameter, the Manager passes the Acquire singleton. Otherwise, the service is instantiated without arguments.

Here is the relevant code from singularity/core/manager.py:

# Inside Manager._register_service_from_path()
service_class = next(
    (cls for name, cls in inspect.getmembers(service_module, inspect.isclass)
     if name.endswith("Service")),
    None,
)

if service_class:
    # Inspect __init__ for 'acquire' parameter
    init_params = inspect.signature(service_class.__init__).parameters
    if "acquire" in init_params:
        service_instance = service_class(acquire=self.acquire)
    else:
        service_instance = service_class()

This approach gives you explicit control: services that need shared resources declare the dependency in their constructor, while lightweight services skip the overhead entirely.

The same pattern applies to middlewares. When the Manager discovers a middleware class in singularity/middleware/, it inspects __init__ for an acquire parameter. The exception handler middleware, for example, uses this to access the WebSocket manager and logger.

Auto-Discovery Mechanics

Auto-discovery is the process by which the Manager finds and registers services without any manual import statements or registration arrays. It relies entirely on filesystem conventions.

Recursive scan -- Starting from services/, the Manager walks every subdirectory. Directories starting with __ (like __base and __pycache__) are skipped.
Disabled check -- If a directory's path appears in .services_disabled, it is skipped. Both full paths (v1/payments) and leaf names (payments) are checked.
Module import -- When a service.py file is found, the Manager constructs the module path (e.g., services.v1.payments.service) and imports it dynamically.
Class lookup -- The Manager uses inspect.getmembers() to find the first class whose name ends with Service.
Signature inspection -- The __init__ signature is checked for an acquire parameter. If present, the Acquire singleton is injected.
Service name injection -- The _service_name attribute is set to the leaf directory name (e.g., payments). This is used by the RPC system to identify callers.
Router creation -- An APIRouter is created with a prefix derived from the directory path (e.g., /api/v1/payments). Core HTTP methods (get, post, put, delete) are registered automatically. Custom routes from http_exposed are registered next.
RPC collection -- If rpc_exposed = True, the service is added to a pending list for RPC registry registration after all services are loaded.

Discovery Path Mapping

The directory structure directly determines the API path and module path:

Directory	Module Path	API Endpoint
`services/auth/service.py`	`services.auth.service`	`/api/auth`
`services/v1/users/service.py`	`services.v1.users.service`	`/api/v1/users`
`services/v1/payments/service.py`	`services.v1.payments.service`	`/api/v1/payments`
`services/v2/users/service.py`	`services.v2.users.service`	`/api/v2/users`

The discovery order depends on the filesystem listing order. Services should not depend on being registered before or after other services. If you need cross-service communication, use the RPC system, which is initialized after all services are registered.

Service Name Injection for RPC

After a service is instantiated, the Manager injects a _service_name attribute set to the leaf directory name:

# Inside Manager._register_service_from_path()
service_instance._service_name = path_segments[-1]

This name serves as the caller identity when making RPC calls. When you call rpc("orders").payments.charge(...), the string "orders" is the caller identity that gets checked against the target service's blacklist. The _service_name injection ensures every service has a consistent, automatically-derived identity without requiring manual configuration.

Code Examples

Service Using Multiple Acquire Resources

A realistic service that combines database access, WebSocket broadcasting, background task submission, and settings:

# services/v1/notifications/service.py
from singularity.core import Service

class NotificationsService(Service):
    def __init__(self, acquire):
        super().__init__(acquire)
        self.db = acquire.db_session
        self.ws = acquire.ws_manager
        self.tasks = acquire.tasks
        self.settings = acquire.settings
        self.logger = acquire.logger

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

        # 2. Push real-time update via WebSocket
        await self.ws.broadcast(
            org_id=user_id,
            data={"type": "notification", "message": message},
            resource="notifications",
        )

        # 3. Queue a follow-up push notification task
        self.tasks.submit("send_push_notification", user_id=user_id)

        return {"status": "sent", "user_id": user_id}

Service Without Acquire

Not every service needs shared resources. A health check endpoint, for example, has no dependencies at all:

# services/health/service.py
from singularity.core import Service

class HealthService(Service):
    async def get(self):
        return {"status": "healthy", "version": "1.0.0"}

The Manager detects that __init__ does not accept acquire and instantiates the class with no arguments. This service still gets a full API route at /api/health with a GET endpoint.

Middleware Using Acquire

Middlewares follow the same injection pattern. The exception handler middleware uses Acquire to access the WebSocket manager and logger:

# singularity/middleware/exceptions.py
from starlette.middleware.base import BaseHTTPMiddleware
from singularity.core.acquire import Acquire

class Middleware(BaseHTTPMiddleware):
    def __init__(self, app, acquire: Acquire):
        super().__init__(app)
        self.ws_manager = acquire.ws_manager
        self.logger = acquire.logger

    async def dispatch(self, request, call_next):
        try:
            return await call_next(request)
        except Exception:
            # Structured error logging and response
            self.logger.exception("Unhandled exception")
            return JSONResponse(status_code=500, content="Internal server error")

Extending Acquire with Custom Dependencies

To add a new shared resource, modify the Acquire class in singularity/core/acquire.py:

class Acquire:
    def __init__(self):
        # ... existing resources ...
        self.stripe = StripeClient(api_key=settings.stripe_api_key)
        self.email = EmailService(smtp_host=settings.smtp_host)

Every service that receives acquire will immediately have access to acquire.stripe and acquire.email with no additional configuration.

Keep Acquire lean. If a dependency is only used by a single service, consider instantiating it directly in that service's __init__ rather than adding it to the global container.

On this page