cosalette¶

An opinionated Python framework for building IoT-to-MQTT bridge applications.

Think "FastAPI for MQTT daemons."

What is cosalette?¶

cosalette provides the common infrastructure that every IoT-to-MQTT bridge needs:

• MQTT lifecycle — connection, reconnection, Last Will and Testament
• Device registration — decorator-based API (@app.command, @app.device, @app.telemetry)
• Configuration — pydantic-settings with environment variables and .env files
• Structured logging — JSON (NDJSON) for production, text for development
• Error reporting — structured errors published to MQTT topics
• Health monitoring — per-device availability and app-level status with LWT
• CLI — --dry-run, --version, --log-level, --log-format, --env-file
• Testing — cosalette.testing module with pytest fixtures and test doubles

Quick Example¶

import cosalette

app = cosalette.App(name="sensor2mqtt", version="0.1.0")

@app.telemetry("sensor", interval=30.0)
async def read_sensor() -> dict[str, object]:
    reading = sensor.read()
    return {"temperature": reading.temp, "humidity": reading.humidity}

if __name__ == "__main__":
    app.run()

For production apps with multiple modules, use Router for composition:

# sensors.py
import cosalette

router = cosalette.Router(prefix="sensors")

@router.telemetry("temperature", interval=30)
async def read_temp() -> dict[str, object]:
    return {"celsius": 22.5}

# main.py
import cosalette
from sensors import router

app = cosalette.App(name="sensor2mqtt", version="1.0.0")
app.include_router(router)
# → publishes to: sensor2mqtt/sensors/temperature/state

Explore the Documentation¶

Architecture Decisions¶

All major design decisions are documented as Architecture Decision Records.

Status¶

cosalette is under active development.

Feedback¶

Was this documentation helpful? We'd love your feedback.

Submit Feedback on GitHub