Dependency Injection Pattern

Overview

All services use constructor-based dependency injection instead of module-level global singletons. Dependencies are created in main() and passed explicitly to consumers via constructors or factory function parameters.

Motivation

• Testability: Tests create fresh instances with mock dependencies directly, no fragile patch() paths
• Explicitness: Each component declares its dependencies in its constructor signature
• Isolation: No shared mutable state between test cases

Pattern

gRPC Services (galaxy, players, physics, tick-engine)

Dependencies flow from main() through factory functions:

main() creates instances
  → create_server(port, deps...) passes to Servicer constructor
  → create_health_app(deps...) passes to health endpoint closures

Servicer Constructor Injection

class PhysicsServicer:
    def __init__(self, redis_state: RedisState, simulation: PhysicsSimulation):
        self._redis_state = redis_state
        self._simulation = simulation

Server Factory

def create_server(port: int, redis_state: RedisState, simulation: PhysicsSimulation) -> grpc.aio.Server:
    server = grpc.aio.server()
    servicer = PhysicsServicer(redis_state, simulation)
    physics_pb2_grpc.add_PhysicsServicer_to_server(servicer, server)
    server.add_insecure_port(f"[::]:{port}")
    return server

Health App Factory (Closure Pattern)

def create_health_app(redis_state: RedisState, simulation: PhysicsSimulation) -> Starlette:
    async def health_ready(request):
        if _shutting_down:
            return JSONResponse({"status": "shutting_down"}, status_code=503)
        if redis_state.connected and simulation.initialized:
            return JSONResponse({"status": "ready"})
        return JSONResponse({"status": "not_ready"}, status_code=503)

    routes = [Route("/health/ready", health_ready), ...]
    return Starlette(routes=routes)

The _shutting_down flag remains module-level in health.py with its set_shutting_down() mutator, since it is set by signal handlers in main() and read by health endpoint closures within the same module.

FastAPI Service (api-gateway)

Dependencies are stored on app.state during startup and accessed via FastAPI’s Depends() system:

Dependency Providers (deps.py)

from fastapi import Request
from .grpc_clients import GrpcClients
from .websocket_manager import WebSocketManager

def get_grpc_clients(request: Request) -> GrpcClients:
    return request.app.state.grpc_clients

def get_ws_manager(request: Request) -> WebSocketManager:
    return request.app.state.ws_manager

Route Handler Injection

@router.post("/api/game/status")
async def game_status(grpc: GrpcClients = Depends(get_grpc_clients)):
    response = await grpc.tick_engine.GetStatus(...)
    return {"status": "ok", ...}

Startup Wiring

@app.on_event("startup")
async def startup():
    app.state.grpc_clients = GrpcClients()
    await app.state.grpc_clients.connect()
    app.state.ws_manager = WebSocketManager(app.state.grpc_clients)
    await app.state.ws_manager.connect()

Encapsulation Rules

Internal attributes (prefixed with _) must not be accessed from outside their owning class. When external code needs access to an internal resource, the class must expose a public property or method.

Example: WebSocketManager.redis

Route handlers need access to the Redis connection managed by WebSocketManager. Rather than accessing ws_manager._redis directly (breaking encapsulation), WebSocketManager exposes a public redis read-only property:

class WebSocketManager:
    def __init__(self, grpc_clients: GrpcClients) -> None:
        self._redis: Optional[redis.Redis] = None
        ...

    @property
    def redis(self) -> Optional[redis.Redis]:
        """Public accessor for the Redis connection."""
        return self._redis

Route handlers use ws_manager.redis instead of ws_manager._redis. This preserves the ability to change the internal storage mechanism without breaking callers.

Service Dependency Maps

Service	Component	Constructor Dependencies
galaxy	GalaxyServicer	GalaxyService
players	PlayersServicer	PlayersService
players	PlayersService	Database
physics	PhysicsServicer	RedisState, PhysicsSimulation
physics	PhysicsSimulation	RedisState
tick-engine	TickEngineServicer	TickEngineState, TickLoop
tick-engine	TickLoop	TickEngineState
api-gateway	WebSocketManager	GrpcClients
api-gateway	Route handlers	via Depends(): GrpcClients, WebSocketManager, AdminAuth, LoginRateLimiter

Testing

With DI, tests construct components directly with mock dependencies:

def test_get_bodies():
    mock_service = MagicMock(spec=GalaxyService)
    mock_service.get_bodies.return_value = []
    servicer = GalaxyServicer(mock_service)
    response = servicer.GetBodies(request, context)
    assert response.bodies == []

No patch() needed for injected dependencies. patch() is still used for non-injected module-level imports (e.g., structlog.get_logger()).