runtime¶
openframe/core/runtime/ · Optional composition root.
Overview¶
ApplicationBootstrap is an optional base class that manages PluginRegistry registration and lifecycle for you. It is an async context manager: __aenter__ calls start() (which calls initialize_all()), and __aexit__ calls stop() (which calls shutdown_all()). If you prefer to manage the registry directly, you do not need ApplicationBootstrap at all.
Classes¶
ApplicationBootstrap¶
class ApplicationBootstrap:
def configure(self) -> None: ...
async def start(self) -> None: ...
async def stop(self) -> None: ...
def register(self, port: BasePort, config: dict | None = None) -> None: ...
def get(self, capability: Capability) -> BasePort: ...
def get_all(self, capability: Capability) -> list[BasePort]: ...
async def __aenter__(self) -> "ApplicationBootstrap": ...
async def __aexit__(self, *args: Any) -> None: ...
configure()¶
Override this method to register ports. Called by start() before initialize_all(). The default implementation is a no-op.
class MyApp(ApplicationBootstrap):
def configure(self) -> None:
self.register(PostgresItemRepository(), config={"dsn": "postgres://..."})
self.register(RedisCache(), config={"url": "redis://..."})
start()¶
Call configure() then registry.initialize_all(). Raises PluginInitializationError if any port fails to initialize.
stop()¶
Call registry.shutdown_all(). Never raises.
register(port, config)¶
Delegate to registry.register(). Raises DuplicatePluginError if the same port instance is registered twice.
get(capability) / get_all(capability)¶
Delegate to registry.get() / registry.get_all(). See plugins module for semantics.
Usage Patterns¶
As an async context manager¶
from openframe.core.runtime import ApplicationBootstrap
from openframe.core.ports import Capability
class MyServiceBootstrap(ApplicationBootstrap):
def configure(self) -> None:
self.register(PostgresItemRepository(), config={"dsn": "..."})
async with MyServiceBootstrap() as bootstrap:
repo = bootstrap.get(Capability.PERSISTENCE)
await serve(ItemService(repo))
# shutdown_all() called automatically on __aexit__
As a FastAPI lifespan¶
from contextlib import asynccontextmanager
from fastapi import FastAPI
bootstrap = MyServiceBootstrap()
@asynccontextmanager
async def lifespan(app: FastAPI):
setup_telemetry()
await bootstrap.start()
yield
await bootstrap.stop()
app = FastAPI(lifespan=lifespan)
Without ApplicationBootstrap¶
ApplicationBootstrap is optional. You can manage PluginRegistry directly: