API Reference¶

def on_configure(self, func: Callable[..., Any]) -> Callable[..., Any]:
    """Register a configuration hook called before devices start.

    The hook runs after settings and adapters are resolved but
    before the run-loop.  Parameters are injected by type
    annotation (Settings, adapter ports, Logger, ClockPort).

    Use ``@app.on_configure`` (no parentheses).

    See Also:
        ADR-023 — on_configure lifecycle phase.
    """
    self._configure_hooks.append(func)
    return func

def device(
    self,
    name: str | None = None,
    *,
    init: Callable[..., Any] | None = None,
    enabled: bool = True,
) -> Callable[..., Any]:
    """Register a command & control device.

    The decorated function runs as a concurrent asyncio task.
    Parameters are injected based on type annotations — declare
    only what you need (e.g. ``ctx: DeviceContext``,
    ``settings: Settings``, ``logger: logging.Logger``).
    Zero-parameter handlers are valid.

    The framework subscribes to ``{name}/set`` and routes commands
    to the handler registered via ``ctx.on_command``.

    When *name* is ``None``, the function name is used internally
    and the device publishes to root-level topics (``{prefix}/state``
    instead of ``{prefix}/{device}/state``).

    Args:
        name: Device name for MQTT topics and logging.  When
            ``None``, the function name is used internally and
            topics omit the device segment.
        init: Optional synchronous factory called once before the
            handler loop.  Its return value is injected into
            the handler by type.
        enabled: When ``False``, registration is silently skipped.
            The decorator returns the original function unmodified
            and no name slot is reserved.  Defaults to ``True``.

    Raises:
        ValueError: If a device with this name is already registered.
        ValueError: If a second root (unnamed) device is registered.
        TypeError: If any handler parameter lacks a type annotation.
    """
    if callable(name) and asyncio.iscoroutinefunction(name):
        raise TypeError("Use @app.device(), not @app.device (parentheses required)")

    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
        if callable(name):
            self.add_device(name, func, init=init, enabled=enabled, is_root=False)
        else:
            resolved_name = name if name is not None else func.__name__
            self.add_device(
                resolved_name,
                func,
                init=init,
                enabled=enabled,
                is_root=name is None,
            )
        return func

    return decorator

def add_device(
    self,
    name: str | Callable[..., Any],
    func: Callable[..., Awaitable[None]],
    *,
    init: Callable[..., Any] | None = None,
    enabled: bool = True,
    is_root: bool = False,
) -> None:
    """Register a command & control device imperatively.

    This is the imperative counterpart to :meth:`device`.  It
    always creates a *named* (non-root) registration by default.

    Args:
        name: Device name for MQTT topics and logging.
        func: Async callable that implements the device loop.
        init: Optional synchronous factory called once before the
            handler loop.  Its return value is injected into
            *func* by type.
        enabled: When ``False``, registration is silently skipped
            — no entry in the registry and no name slot reserved.
            Defaults to ``True``.
        is_root: When ``True``, the device publishes to root-level
            topics (``{prefix}/state`` instead of
            ``{prefix}/{name}/state``).  Defaults to ``False``.

    Raises:
        ValueError: If a device with this name is already registered.
        TypeError: If *init* is async or has un-annotated parameters.
        TypeError: If *func* has un-annotated parameters.

    See Also:
        :meth:`device` — decorator equivalent.
    """
    if not enabled:
        return
    if init is not None:
        _validate_init(init)
    init_plan = build_injection_plan(init) if init is not None else None
    if not callable(name):
        check_device_name(
            name,
            registry_type="device",
            is_root=is_root,
            devices=self._devices,
            telemetry=self._telemetry,
            commands=self._commands,
        )
    plan = build_injection_plan(func)
    if callable(name):
        self._devices.append(
            _DeviceRegistration(
                name=func.__qualname__,
                func=func,
                injection_plan=plan,
                is_root=is_root,
                init=init,
                init_injection_plan=init_plan,
                name_spec=name,
            ),
        )
    else:
        self._devices.append(
            _DeviceRegistration(
                name=name,
                func=func,
                injection_plan=plan,
                is_root=is_root,
                init=init,
                init_injection_plan=init_plan,
            ),
        )

def command(
    self,
    name: str | None = None,
    *,
    init: Callable[..., Any] | None = None,
    enabled: bool = True,
) -> Callable[..., Any]:
    """Register a command handler for an MQTT device.

    The decorated function is called each time a command arrives
    on the ``{prefix}/{name}/set`` topic.  Parameters named
    ``topic`` and ``payload`` receive the MQTT message values;
    all other parameters are injected by type annotation, exactly
    like ``@app.device`` and ``@app.telemetry`` handlers.

    If the handler returns a ``dict``, the framework publishes it
    as device state via ``publish_state()``.  Return ``None`` to
    skip auto-publishing.

    When *name* is ``None``, the function name is used internally
    and the device publishes to root-level topics.

    Args:
        name: Device name used for MQTT topics and logging.  When
            ``None``, the function name is used internally and
            topics omit the device segment.
        init: Optional synchronous factory called once before the
            handler loop.  Its return value is injected into
            the handler by type.
        enabled: When ``False``, registration is silently skipped.
            The decorator returns the original function unmodified
            and no name slot is reserved.  Defaults to ``True``.

    Raises:
        ValueError: If a device with this name is already registered.
        ValueError: If a second root (unnamed) device is registered.
        TypeError: If any handler parameter lacks a type annotation.
    """
    if callable(name) and asyncio.iscoroutinefunction(name):
        raise TypeError(
            "Use @app.command(), not @app.command (parentheses required)"
        )

    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
        if callable(name):
            self.add_command(name, func, init=init, enabled=enabled, is_root=False)
        else:
            resolved_name = name if name is not None else func.__name__
            self.add_command(
                resolved_name,
                func,
                init=init,
                enabled=enabled,
                is_root=name is None,
            )
        return func

    return decorator

def add_command(
    self,
    name: str | Callable[..., Any],
    func: Callable[..., Awaitable[dict[str, object] | None]],
    *,
    init: Callable[..., Any] | None = None,
    enabled: bool = True,
    is_root: bool = False,
) -> None:
    """Register a command handler imperatively.

    This is the imperative counterpart to :meth:`command`.  It
    always creates a *named* (non-root) registration by default.

    Args:
        name: Device name for MQTT topics and logging.
        func: Async callable invoked on each incoming command.
            Parameters named ``topic`` and ``payload`` receive the
            MQTT message values; others are injected by type.
        init: Optional synchronous factory called once before the
            handler loop.  Its return value is injected into
            *func* by type.
        enabled: When ``False``, registration is silently skipped
            — no entry in the registry and no name slot reserved.
            Defaults to ``True``.
        is_root: When ``True``, the device publishes to root-level
            topics (``{prefix}/state`` instead of
            ``{prefix}/{name}/state``).  Defaults to ``False``.

    Raises:
        ValueError: If a device with this name is already registered.
        TypeError: If *init* is async or has un-annotated parameters.
        TypeError: If *func* has un-annotated parameters.

    See Also:
        :meth:`command` — decorator equivalent.
    """
    if not enabled:
        return
    if init is not None:
        _validate_init(init)
    init_plan = build_injection_plan(init) if init is not None else None
    if not callable(name):
        check_device_name(
            name,
            registry_type="command",
            is_root=is_root,
            devices=self._devices,
            telemetry=self._telemetry,
            commands=self._commands,
        )
    plan = build_injection_plan(func, mqtt_params={"topic", "payload"})
    sig = inspect.signature(func)
    declared_mqtt = frozenset({"topic", "payload"} & sig.parameters.keys())
    if callable(name):
        self._commands.append(
            _CommandRegistration(
                name=func.__qualname__,
                func=func,
                injection_plan=plan,
                mqtt_params=declared_mqtt,
                is_root=is_root,
                init=init,
                init_injection_plan=init_plan,
                name_spec=name,
            ),
        )
    else:
        self._commands.append(
            _CommandRegistration(
                name=name,
                func=func,
                injection_plan=plan,
                mqtt_params=declared_mqtt,
                is_root=is_root,
                init=init,
                init_injection_plan=init_plan,
            ),
        )

def telemetry(
    self,
    name: str | None = None,
    *,
    interval: IntervalSpec,
    publish: PublishStrategy | None = None,
    persist: PersistPolicy | None = None,
    init: Callable[..., Any] | None = None,
    enabled: bool = True,
    group: str | None = None,
    retry: int = 0,
    retry_on: tuple[type[BaseException], ...] | None = None,
    backoff: BackoffStrategy | None = None,
    circuit_breaker: CircuitBreaker | None = None,
) -> Callable[..., Any]:
    """Register a telemetry device with periodic polling.

    The decorated function returns a ``dict`` published as JSON
    state, or ``None`` to suppress publishing for that cycle.
    Parameters are injected based on type annotations — declare
    only what you need.  Zero-parameter handlers are valid.

    The framework calls the handler at the specified interval
    and publishes the returned dict (unless suppressed by a
    ``None`` return or a publish strategy).

    When *name* is ``None``, the function name is used internally
    and the device publishes to root-level topics.

    Args:
        name: Device name for MQTT topics and logging.  When
            ``None``, the function name is used internally and
            topics omit the device segment.
        interval: Polling interval in seconds, or a callable
            ``(Settings) -> float`` for deferred resolution.
            When a callable is provided, it is invoked once in
            :meth:`_run_async` after settings are resolved —
            this allows reading intervals from settings without
            requiring valid settings at registration time (e.g.
            during ``--help`` / ``--version``).
        publish: Optional publish strategy controlling when
            readings are actually published (e.g. ``OnChange()``,
            ``Every(seconds=60)``).  When ``None``, every reading
            is published unconditionally.
        persist: Optional save policy controlling when the
            :class:`DeviceStore` is persisted (e.g.
            ``SaveOnPublish()``, ``SaveOnChange()``).  Requires
            ``store=`` on the :class:`App`.  When ``None``, the
            store is saved only on shutdown (the safety net).
        init: Optional synchronous factory called once before the
            handler loop.  Its return value is injected into
            the handler by type.
        enabled: When ``False``, registration is silently skipped.
            The decorator returns the original function unmodified
            and no name slot is reserved.  Defaults to ``True``.
        group: Optional coalescing group name.  Telemetry devices
            in the same group share a single scheduler tick so
            their readings are published together.  When ``None``
            (the default), the device runs on its own independent
            timer.
        retry: Maximum number of retry attempts after a failure.
            Defaults to ``0`` (no retry).  The retry counter
            persists across poll cycles and resets on success.
        retry_on: Exception types to retry on.  Defaults to
            ``(OSError,)`` when ``retry > 0`` and not explicitly
            set.  Exceptions not matching this tuple propagate
            immediately to the error handler.
        backoff: Backoff strategy controlling delay between retries
            (e.g. ``ExponentialBackoff()``, ``LinearBackoff()``,
            ``FixedBackoff()``).  Defaults to
            ``ExponentialBackoff(base=2.0, max_delay=60.0)`` when
            ``retry > 0`` and not explicitly set.
        circuit_breaker: Optional circuit breaker that stops
            retrying after consecutive failed cycles.  Works
            independently of ``retry`` — even with ``retry=0``,
            it tracks per-cycle failures.

    Raises:
        ValueError: If a device with this name is already registered.
        ValueError: If a second root (unnamed) device is registered.
        ValueError: If *interval* is a float and <= 0.  For
            callable intervals, validation is deferred to
            :meth:`_run_async`.
        ValueError: If ``persist`` is set but no ``store=`` backend
            was configured on the App.
        ValueError: If *group* is an empty string.
        ValueError: If ``retry > 0`` and ``retry_on`` is
            explicitly empty.
        TypeError: If any handler parameter lacks a type annotation.
    """
    # Skip all validation when disabled — a disabled device shouldn't raise.
    if enabled and group is not None and group == "":
        msg = "group must be non-empty"
        raise ValueError(msg)

    # Eagerly validate persist/store at decoration time
    # (add_telemetry re-checks for the imperative path).
    # Skip when disabled — a disabled device shouldn't raise.
    if enabled and persist is not None and self._store is None:
        msg = (
            "persist= requires a store= backend on the App. "
            "Pass store=MemoryStore() (or another Store) to App()."
        )
        raise ValueError(msg)

    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
        if callable(name):
            self.add_telemetry(
                name,
                func,
                interval=interval,
                publish=publish,
                persist=persist,
                init=init,
                enabled=enabled,
                group=group,
                is_root=False,
                retry=retry,
                retry_on=retry_on,
                backoff=backoff,
                circuit_breaker=circuit_breaker,
            )
        else:
            resolved_name = name if name is not None else func.__name__
            self.add_telemetry(
                resolved_name,
                func,
                interval=interval,
                publish=publish,
                persist=persist,
                init=init,
                enabled=enabled,
                group=group,
                is_root=name is None,
                retry=retry,
                retry_on=retry_on,
                backoff=backoff,
                circuit_breaker=circuit_breaker,
            )
        return func

    return decorator

def add_telemetry(
    self,
    name: str | Callable[..., Any],
    func: Callable[..., Awaitable[dict[str, object] | None]],
    *,
    interval: IntervalSpec,
    publish: PublishStrategy | None = None,
    persist: PersistPolicy | None = None,
    init: Callable[..., Any] | None = None,
    enabled: bool = True,
    group: str | None = None,
    is_root: bool = False,
    retry: int = 0,
    retry_on: tuple[type[BaseException], ...] | None = None,
    backoff: BackoffStrategy | None = None,
    circuit_breaker: CircuitBreaker | None = None,
) -> None:
    """Register a telemetry device imperatively.

    This is the imperative counterpart to :meth:`telemetry`.  It
    always creates a *named* (non-root) registration by default.

    Args:
        name: Device name for MQTT topics and logging.
        func: Async callable returning a ``dict`` (published as
            state) or ``None`` (suppresses that cycle).
        interval: Polling interval in seconds, or a callable
            ``(Settings) -> float`` for deferred resolution.
            When a callable is provided, it is invoked once in
            :meth:`_run_async` after settings are resolved —
            this allows reading intervals from settings without
            requiring valid settings at registration time (e.g.
            during ``--help`` / ``--version``).
        publish: Optional publish strategy (e.g. ``OnChange()``)
            controlling when readings are actually published.
        persist: Optional save policy.  Requires ``store=`` on the
            :class:`App`.
        init: Optional synchronous factory called once before the
            handler loop.  Its return value is injected into
            *func* by type.
        enabled: When ``False``, registration is silently skipped
            — no entry in the registry and no name slot reserved.
            Defaults to ``True``.
        group: Optional coalescing group name.  Telemetry devices
            in the same group share a single scheduler tick so
            their readings are published together.  When ``None``
            (the default), the device runs on its own independent
            timer.
        is_root: When ``True``, the device publishes to root-level
            topics (``{prefix}/state`` instead of
            ``{prefix}/{name}/state``).  Defaults to ``False``.

    Raises:
        ValueError: If a device with this name is already registered.
        ValueError: If *interval* is a float and <= 0.  For
            callable intervals, validation is deferred to
            :meth:`_run_async`.
        ValueError: If *persist* is set but no ``store=`` backend
            was configured on the App.
        ValueError: If *group* is an empty string.
        TypeError: If *init* is async or has un-annotated parameters.
        TypeError: If *func* has un-annotated parameters.

    See Also:
        :meth:`telemetry` — decorator equivalent.
    """
    if not enabled:
        return
    self._validate_telemetry_args(
        name,
        interval,
        persist,
        init,
        group,
        retry=retry,
        retry_on=retry_on,
    )
    init_plan = build_injection_plan(init) if init is not None else None
    if not callable(name):
        check_device_name(
            name,
            registry_type="telemetry",
            is_root=is_root,
            devices=self._devices,
            telemetry=self._telemetry,
            commands=self._commands,
        )
    plan = build_injection_plan(func)
    resolved_name = func.__qualname__ if callable(name) else name
    name_spec = name if callable(name) else None

    # Resolve retry defaults
    resolved_retry_on = retry_on
    resolved_backoff = backoff
    if retry > 0:
        if resolved_retry_on is None:
            resolved_retry_on = _DEFAULT_RETRY_ON
        if resolved_backoff is None:
            resolved_backoff = _DEFAULT_BACKOFF

    self._telemetry.append(
        _TelemetryRegistration(
            name=resolved_name,
            func=func,
            injection_plan=plan,
            interval=interval,
            is_root=is_root,
            publish_strategy=publish,
            persist_policy=persist,
            init=init,
            init_injection_plan=init_plan,
            group=group,
            name_spec=name_spec,
            retry=retry,
            retry_on=resolved_retry_on if resolved_retry_on is not None else (),
            backoff=resolved_backoff,
            circuit_breaker=circuit_breaker,
        ),
    )

def adapter(
    self,
    port_type: type,
    impl: type | str | Callable[..., object],
    *,
    dry_run: type | str | Callable[..., object] | None = None,
) -> None:
    """Register an adapter for a port type.

    All adapter forms support dependency injection: if a class
    ``__init__`` or factory callable declares a parameter
    annotated with ``Settings`` (or a subclass), the parsed
    settings instance is auto-injected at resolution time.

    Args:
        port_type: The Protocol type to register.
        impl: The adapter class, a ``module:ClassName`` lazy import
            string, or a factory callable returning an adapter instance.
        dry_run: Optional dry-run variant (class, lazy import string,
            or factory callable).

    Raises:
        ValueError: If an adapter is already registered for this port type.
        TypeError: If a callable (class or factory) has invalid
            signatures (e.g. un-annotated parameters or
            unresolvable types).
    """
    if port_type in self._adapters:
        msg = f"Adapter already registered for {port_type!r}"
        raise ValueError(msg)

    # Fail-fast: validate callable signatures at registration time
    # so errors surface here rather than at runtime resolution.
    # Classes are included — inspect.signature(cls) inspects __init__.
    for candidate in (impl, dry_run):
        if (
            candidate is not None
            and callable(candidate)
            and not isinstance(candidate, str)
        ):
            build_injection_plan(candidate)

    self._adapters[port_type] = _AdapterEntry(impl=impl, dry_run=dry_run)

def run(
    self,
    *,
    mqtt: MqttPort | None = None,
    settings: Settings | None = None,
    shutdown_event: asyncio.Event | None = None,
    clock: ClockPort | None = None,
) -> None:
    """Start the application (blocking, synchronous entrypoint).

    Wraps :meth:`_run_async` in :func:`asyncio.run`, handling
    ``KeyboardInterrupt`` for clean Ctrl-C shutdown.  This is the
    recommended way to launch a cosalette application::

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

    All parameters are optional and intended for programmatic or
    test use — production apps typically call ``run()`` with no
    arguments.

    Args:
        mqtt: Override MQTT client (e.g. ``MockMqttClient`` for
            testing).  When ``None``, a real ``MqttClient`` is
            created from settings.
        settings: Override settings (skip env-file loading).
        shutdown_event: Override shutdown event (skip OS signal
            handlers).  Useful in tests to control shutdown timing.
        clock: Override clock (e.g. ``FakeClock`` for tests).

    See Also:
        :meth:`cli` — CLI entrypoint with Typer argument parsing.
    """
    with contextlib.suppress(KeyboardInterrupt):
        asyncio.run(
            self._run_async(
                mqtt=mqtt,
                settings=settings,
                shutdown_event=shutdown_event,
                clock=clock,
            ),
        )

def cli(self) -> None:
    """Start the application with CLI argument parsing.

    Builds a Typer CLI from the application's configuration,
    parses command-line arguments (``--dry-run``, ``--version``,
    ``--log-level``, ``--log-format``, ``--env-file``), and
    orchestrates the full async lifecycle.

    For production use without CLI parsing, prefer :meth:`run`.

    See Also:
        ADR-005 — CLI framework.
    """
    from cosalette._cli import build_cli

    cli = build_cli(self)
    cli(standalone_mode=True)