BootstrapProtocol
Protocol defining the interface for bootstrap steps.
Bootstrap steps run during application startup in order (sorted by `order` field).
Each step can initialize services, bind adapters to the DI container, or
perform other startup tasks.
Properties:
name: Unique identifier for this bootstrap step
order: Execution order (lower runs first, e.g., 10 before 20)
Methods:
run: Execute the bootstrap step
Example - Standard SQLAlchemy bootstrap:
class InitEngine:
name = "init_engine"
order = 10
async def run(self, container: Container) -> None:
from sqlalchemy.ext.asyncio import create_async_engine
engine = create_async_engine("postgresql+asyncpg://...")
container.bind(Engine, engine)
class SyncSchema:
name = "sync_schema"
order = 30 # Runs after init_engine (10) and init_registries (20)
async def run(self, container: Container) -> None:
mapper = container.get(SchemaMapper)
registry = container.get(MetaRegistry)
# Create tables for all registered DocTypes
for doctype in registry.get_all():
await mapper.create_table(doctype, metadata)
Example - MX MongoDB bootstrap:
class InitMongoClient:
name = "init_mongo"
order = 10
async def run(self, container: Container) -> None:
from motor.motor_asyncio import AsyncIOMotorClient
client = AsyncIOMotorClient("mongodb://localhost")
db = client["my_database"]
container.bind(MongoClient, client)
container.bind(Database, db)
Entry Point Registration (pyproject.toml):
[project.entry-points."framework_m.bootstrap"]
init_engine = "framework_m_standard.bootstrap:InitEngine"
sync_schema = "framework_m_standard.bootstrap:SyncSchema"
# MX package can override or add steps:
init_mongo = "framework_mx_mongo.bootstrap:InitMongoClient"
Source: bootstrap.py
Methods
run
async def run(self, container: Any) -> None
Execute the bootstrap step.
Args:
container: DI container for binding services
(dependency-injector Container or compatible)
Raises:
Exception: If bootstrap step fails critically
Example:
class InitEngine:
name = "init_engine"
order = 10
async def run(self, container: Container) -> None:
engine = create_async_engine(os.getenv("DATABASE_URL"))
container.bind(Engine, engine)
print(f"✓ Database engine initialized")