Skip to main content

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")