Plugins System

Base Petal

class petal_app_manager.plugins.base.Petal

Bases: ABC

Petal authors only inherit this; NO FastAPI import, no routers.

async async_shutdown() → None

Called before shutdown to handle async operations like MQTT unsubscriptions.

async async_startup() → None

Called after startup to handle async operations like MQTT subscriptions.

has_mqtt_actions() → bool

Return True if any method is decorated with @mqtt_action.

inject_proxies(proxies: Mapping[str, BaseProxy]) → None

inject_templates(templates: Mapping[str, Jinja2Templates]) → None

Inject Jinja2 templates into the petal.

name: str

shutdown() → None

Called when the petal is stopped.

startup() → None

Called when the petal is started.

version: str

Decorators

petal_app_manager.plugins.decorators.http_action(method: Literal['GET', 'POST'], path: str, **kwargs)

Marks a petal method as an HTTP endpoint.

petal_app_manager.plugins.decorators.mqtt_action(command: str, *, cpu_heavy: bool = False, **kwargs)

Marks a petal method as an MQTT command handler.

The decorated method must be async def and accept (self, topic: str, message: dict)—exactly the same signature that the legacy _command_handlers dict values have today.

Parameters:

• command (str) – The suffix of the MQTT command, for example "mission_plan" or "fetch_flight_records". At runtime the framework prefixes the petal’s name attribute automatically, producing "petal-leafsdk/mission_plan" etc.

• cpu_heavy (bool) – If True the entire handler invocation will be offloaded to the proxy’s thread-pool executor so CPU-intensive work (e.g. NumPy) does not block the event loop. Defaults to False.

petal_app_manager.plugins.decorators.websocket_action(path: str, **kwargs)

Marks a petal method as a WebSocket endpoint.

Loader

petal_app_manager.plugins.loader.initialize_petals(petal_name_list: List[str], proxies: Dict[str, BaseProxy], logger: Logger) → List[Petal]

Initialize petals without starting them up. Loads petal classes, instantiates them, and injects proxies. Does NOT call petal.startup().

Returns a list of initialized (but not started) Petal objects.

petal_app_manager.plugins.loader.load_petals(app: FastAPI, petal_name_list: List[str], proxies: Dict[str, BaseProxy], logger: Logger) → List[Petal]

Load, initialize, and start petals (convenience function). This is a wrapper that calls initialize_petals() and startup_petals().

For more control over the initialization and startup process, use initialize_petals() and startup_petals() separately.

petal_app_manager.plugins.loader.startup_petals(app: FastAPI, petal_list: List[Petal], logger: Logger) → List[Petal]

Start up initialized petals and mount them to the FastAPI app. Calls petal.startup() and mounts static files, templates, and routers.

Returns the list of successfully started petals.