Router and Composition¶

The Router class lets you organize related devices, commands, and telemetry into separate modules without circular import dependencies. It's the cosalette equivalent of FastAPI's APIRouter — a composition primitive that makes multi-module applications clean and testable.

When to Use Router¶

Key benefits of Router:

• Module independence — Define devices in separate files without importing App
• Testable boundaries — Unit test router modules in isolation
• Topic prefixing — Group related devices under a common MQTT topic segment
• Tag accumulation — Apply metadata (e.g., ["environment"], ["production"]) at multiple layers
• Adapter scoping — Register adapters at the router level

Basic Usage¶

Create a router in a separate module, register devices on it, then include it in your app's composition root:

import cosalette

router = cosalette.Router(prefix="sensors", tags=["environment"])


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


@router.command("calibrate")
async def calibrate(payload: str, ctx: cosalette.DeviceContext) -> dict[str, object]:
    """Calibrate the temperature sensor."""
    # Calibration logic here
    return {"calibrated": True}

import cosalette
from sensors import router as sensors_router

app = cosalette.App(name="home2mqtt", version="1.0.0")
app.include_router(sensors_router)

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

MQTT topics:

• Telemetry: home2mqtt/sensors/temperature/state
• Command: home2mqtt/sensors/calibrate/set

The prefix="sensors" parameter adds a topic segment between the app name and the device name.

Router Constructor Parameters¶

router = cosalette.Router(
    prefix="sensors",            # Single MQTT topic segment (no `/` allowed)
    tags=["environment", "production"],  # Metadata for introspection
    adapters={...},              # Adapter registrations (see below)
)

Prefix validation:

• Must be a single MQTT topic segment (no / character, no wildcards)
• Validated by the same rules as device names (ADR-002)

Router Decorators¶

Router provides the same decorator surface as App:

• @router.telemetry(name, interval=...)
• @router.command(name)
• @router.device(name)
• @router.stream(name)
• @router.periodic(interval=...)
• @router.react(state_type)

Router decorators extend their @app.* equivalents in two ways:

1. tags= per operation — @router.telemetry, @router.command, @router.device, @router.stream, and @router.periodic each accept an additional tags= keyword argument. Tags declared at the operation level accumulate with the router constructor's tags and include_router tags (see Tag Accumulation).

2. Deferred state_type validation in Router.react — App.react validates that state_type is already registered via @app.state and raises ValueError immediately if not. Router.react defers this check to include_router call time, because the router has no access to the app's state factories before inclusion.

All other parameters, runtime behavior, and semantics are identical to their @app.* equivalents. See the respective guides for usage patterns.

Including a Router¶

Use App.include_router() to merge router registrations into your app:

app.include_router(
    router,
    prefix="sensors",         # Additional prefix level (optional)
    tags=["production"],      # Additional tags (optional)
    adapters={...},           # Additional adapters (optional)
)

Prefix combination:

Example with two prefix levels:

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

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

app.include_router(router, prefix="sensors")
# → Topic: myapp/sensors/temp/outside/state

Tag Accumulation¶

Tags accumulate from three sources, deduplicated while preserving insertion order:

1. Router constructor (tags=["environment"])
2. include_router() call (tags=["production"])
3. Operation decorator (@router.telemetry(..., tags=["critical"]))

router = cosalette.Router(prefix="sensors", tags=["environment"])

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

app.include_router(router, tags=["production"])

# Final tags on the registration: ["environment", "production", "critical"]

Tags are introspection metadata — they appear in app.asyncapi() output and can be queried by MCP tools or monitoring systems. They have no runtime effect on MQTT behavior.

Adapter Scoping¶

Routers can declare their own adapter registrations:

from sensors.ports import TemperatureSensorPort
from sensors.adapters import I2CTemperatureSensor

router = cosalette.Router(
    prefix="sensors",
    adapters={
        TemperatureSensorPort: I2CTemperatureSensor,
    },
)


@router.telemetry("temperature", interval=30)
async def read_temperature(ctx: cosalette.DeviceContext) -> dict[str, object]:
    sensor = ctx.adapter(TemperatureSensorPort)
    return {"celsius": sensor.read()}

Adapters from Router(adapters={...}) and include_router(adapters={...}) are merged into the app's adapter registry. Conflicts (same port registered twice with different implementations) raise ValueError at include_router() call time.

Testing Routers¶

Routers can be tested independently without instantiating an App:

import pytest
from sensors import router


def test_router_has_temperature_device() -> None:
    """Verify temperature telemetry is registered."""
    assert "temperature" in router.registered_names

Asserting registration metadata¶

registered_names answers "is this name registered?" but gives no metadata. The six collection properties expose the registration objects directly — including fields such as payload_model, state_model, init, and scheduling metadata — and mirror the equivalent properties on App:

from myapp.devices.system import router, SystemActionCommand, SystemActionState


def test_action_command_metadata() -> None:
    """The action command wires the expected payload and state models."""
    reg = next(r for r in router.commands if r.name == "action")
    assert reg.payload_model is SystemActionCommand
    assert reg.state_model is SystemActionState

All five sequence properties (commands, telemetry_registrations, devices, periodic_registrations, stream_registrations) return immutable point-in-time snapshots (tuple), so router internals cannot be mutated through them. The adapters property returns a MappingProxyType — an immutable but live view of the adapter registry; entries added after obtaining the proxy remain visible through it. The _registrations suffix on telemetry_registrations and periodic_registrations avoids shadowing the @router.telemetry and @router.periodic decorators.

For integration tests, use AppHarness with include_router():

from cosalette.testing import AppHarness
from sensors import router


async def test_temperature_telemetry(harness: AppHarness) -> None:
    """Integration test for temperature telemetry via router."""
    harness.app.include_router(router)

    await harness.run()
    await harness.advance_time(30)

    harness.assert_published("testapp/sensors/temperature/state", contains="celsius")

See the Testing guide for full harness patterns.

Snapshot Semantics¶

include_router() captures registrations at call time — modifications to the router after inclusion do not affect the app:

router = cosalette.Router()

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

app.include_router(router)  # Snapshot taken here

# This registration is NOT included in the app
@router.telemetry("humidity", interval=30)
async def humidity() -> dict[str, object]:
    return {"percent": 55.0}

This matches FastAPI's behavior and makes composition order explicit.

Multiple Inclusion¶

A router can be included multiple times, optionally with different prefixes or tags:

router = cosalette.Router(tags=["sensor"])

@router.telemetry("reading", interval=30)
async def reading() -> dict[str, object]:
    return {"value": 42}

app.include_router(router, prefix="indoor", tags=["environment"])
app.include_router(router, prefix="outdoor", tags=["environment"])

# Two registrations created:
# - myapp/indoor/reading/state with tags: ["sensor", "environment"]
# - myapp/outdoor/reading/state with tags: ["sensor", "environment"]

Each inclusion creates independent device contexts and task groups.

Nested Routers (Not Supported)¶

Router.include_router() does not exist — routers cannot include other routers. Multi-level composition must be done at the App level:

# WRONG — Router has no include_router method
parent_router.include_router(child_router)  # AttributeError

# RIGHT — App composes multiple routers
app.include_router(sensors_router, prefix="env")
app.include_router(controls_router, prefix="cmd")

This constraint is documented in ADR-044 and may be revisited in a future release based on user feedback.

When to Use App-Level vs Router¶

Use app-level decorators when:

• Building a quickstart example or tutorial project
• Single-file application (≤200 lines, ≤3 devices)
• Prototyping or exploratory development
• No shared libraries or reusable modules needed

Use Router when:

• Multi-file production application
• Separating devices by domain (sensors, controls, diagnostics)
• Building a library of reusable device groups
• Testing device logic independently from app wiring

Hybrid approach:

You can mix both patterns in the same app — use app-level decorators for singleton devices and routers for grouped modules:

from sensors import router as sensors_router

app = cosalette.App(name="home2mqtt", version="1.0.0")

# Router for grouped sensors
app.include_router(sensors_router)

# App-level decorator for a singleton heartbeat
@app.telemetry("heartbeat", interval=60)
async def heartbeat() -> dict[str, object]:
    return {"uptime_seconds": get_uptime()}

See Also¶

• ADR-044: Public Router and Composition API — Full design rationale
• Architecture — Composition root pattern
• Testing — Testing routers with AppHarness
• Contract-First Route Design — Typed router operations