Skip to main content

​
Application Lifecycle & Dependency Management

​
Overview

The application uses Reflex’s register_lifespan_task API to run startup validation and cleanup logic. This mirrors FastAPI’s lifespan pattern using @asynccontextmanager, adapted for the Reflex framework. All lifecycle and dependency-injection logic lives in app/api/deps.py, keeping app/app.py as a thin wiring layer.

​
Architecture

┌─────────────────────────────────────────────────────────────────┐
│                         app/app.py                              │
│                    (thin wiring layer)                          │
│                                                                 │
│  fastapi_app = FastAPI(...)          # REST API                 │
│  fastapi_app.include_router(health)  # /health endpoint        │
│                                                                 │
│  app = rx.App(api_transformer=fastapi_app)                      │
│  app.register_lifespan_task(app_lifespan)   ◄── from deps.py   │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                       app/api/deps.py                           │
│              (lifecycle + dependency injection)                  │
│                                                                 │
│  @asynccontextmanager                                           │
│  async def app_lifespan(app: FastAPI):                          │
│      # STARTUP                                                  │
│      1. Validate Hydra config (resolve ${oc.env:…} early)       │
│      2. Register services → ServiceRegistry                     │
│      3. Initialize all services (LLM chain, chat)               │
│      4. Log health check results                                │
│      yield  ──── application serves requests ────               │
│      # SHUTDOWN                                                 │
│      5. Reset LLM client singleton                              │
│                                                                 │
│  def get_llm_service() → LLMService      # Depends() callable  │
│  def get_chat_service() → ChatService     # Depends() callable  │
└─────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                   app/api/services/                              │
│                                                                 │
│  registry.py   → ServiceRegistry (register, get, health_check)  │
│  base.py       → BaseService ABC (initialize, health_check)     │
│  llm_service.py → LLMService (wraps LLMClientManager)           │
│  chat_service.py → ChatService (depends on LLM availability)    │
└─────────────────────────────────────────────────────────────────┘

​
Startup Sequence

​
1. Config Validation

Loads get_app_config() and get_llm_config() from app/hydra_config.py. This forces Hydra to resolve all ${oc.env:...} references, surfacing missing environment variables immediately rather than on first user request. Logs the resolved app name, database URL (truncated), LLM model, and provider.

​
2. Service Registration

Calls _register_services() which populates ServiceRegistry: 
ServiceRegistry.register("llm", LLMService())
ServiceRegistry.register("chat", ChatService())

​
3. Service Initialization

Calls ServiceRegistry.initialize_all() which invokes initialize() on each registered service:
  • LLMService — calls LLMClientManager.get_client() to eagerly build the multi-provider fallback chain. Validates API keys and provider availability at boot time.
  • ChatService — checks that its LLM dependency is available.

​
4. Health Check

Calls ServiceRegistry.health_check_all() and logs results. This provides observability into which services started successfully.

​
Shutdown Cleanup

Calls LLMClientManager.reset() to release cached LLM connections and provider resources.

​
Service Registry

The ServiceRegistry (app/api/services/registry.py) is a class-level registry for managing service lifecycle:
MethodPurpose
register(name, service)Register a BaseService instance
get(name)Retrieve a service by name (raises KeyError if missing)
initialize_all()Call initialize() on all registered services
health_check_all()Run health_check() on all services, return {name: bool}
list_services()List registered service names

​
Adding a New Service

  1. Create a class extending BaseService in app/api/services/:
from .base import BaseService

class MyService(BaseService):
    async def initialize(self) -> None:
        # setup logic
        self._available = True

    async def health_check(self) -> bool:
        return self._available
  1. Register it in _register_services() in app/api/deps.py:
ServiceRegistry.register("my_service", MyService())
  1. Add a Depends() callable if needed by FastAPI routes:
def get_my_service() -> MyService:
    return ServiceRegistry.get("my_service")

​
Dependency Injection

​
FastAPI Routes

Use Depends() callables from app/api/deps.py: 
from fastapi import Depends
from app.api.deps import get_llm_service
from app.api.services.llm_service import LLMService

@router.post("/chat")
async def chat(llm: LLMService = Depends(get_llm_service)):
    client = llm.get_client()
    # ...

​
Reflex State Handlers

Use rx.session() (synchronous) for database access, managed by Reflex’s internal session factory configured via db_url in rxconfig.py: 
with rx.session() as session:
    result = session.exec(select(MyModel).where(...)).all()
Database connectivity is validated by Reflex at startup — no separate check is needed in the lifespan.

​
Configuration Source

All configuration flows through the Hydra system: 
envs/{APP_ENV}          → environment variables (loaded by hydra_config.py)
app/config/*.yaml       → Hydra YAML configs (resolved with ${oc.env:...})
app/config/schemas/*.py → typed dataclass schemas
app/hydra_config.py     → accessor functions (get_app_config, get_llm_config, etc.)
app/config.py           → backward-compatible re-export of hydra_config
Access config anywhere in the app: 
from app.hydra_config import get_app_config, get_llm_config

app_config = get_app_config()
db_url = app_config.database.url
llm_model = get_llm_config().model

​
Reflex Lifespan API

Reflex wraps FastAPI internally. The register_lifespan_task method accepts an @asynccontextmanager function: 
from contextlib import asynccontextmanager

@asynccontextmanager
async def my_lifespan(app):
    # startup code
    yield
    # shutdown code

app = rx.App(...)
app.register_lifespan_task(my_lifespan)
Multiple lifespan tasks can be registered — they run in registration order on startup and reverse order on shutdown.

​
Testing

Override dependencies in tests by patching the lifespan or resetting services: 
def test_with_mock_llm():
    LLMClientManager.reset()
    # ... test code that will re-initialise on next get_client() call
The lifespan checks are non-fatal (log errors, don’t crash), so the test suite works even without a running database or API keys.