Schema Enforcement¶

cosalette apps publish to predictable MQTT topics by construction — the framework sets {prefix}/{device}/state, {prefix}/{device}/set, and {prefix}/{device}/availability automatically. But the payload shape and the cross-app contract (which topics a fleet expects to exist) are not enforced.

Schema enforcement fills that gap using AsyncAPI 3.0.0 documents annotated with x-cosalette-* extensions. The key benefits:

• Catch regressions before deployment. A renamed field in one app silently breaks consumers in another. A schema validation step in your CI/CD pipeline or deploy scripts catches it before the app reaches the broker.
• Machine-readable contract. Monitoring tools, code generators, and dashboards can discover which topics your fleet produces and what payloads to expect.
• Zero friction when unused. The default enforcement mode is off — no new dependencies, no new topics, no broker configuration.

pip install cosalette[schema]

Quick Start¶

1 — Generate a starter schema from your app¶

# Scaffold a schema with x-cosalette extensions included
cosalette schema init --app myapp.app:app > schema.yaml

init introspects the running app and produces an AsyncAPI 3.0.0 document:

asyncapi: 3.0.0
info:
  title: thermo2mqtt
  version: 0.1.0

x-cosalette-enforcement:
  mode: warn
  on_configure: true
  on_publish: false
  network_level: false

channels:
  temperatureState:
    address: thermo2mqtt/temperature/state
    x-cosalette-archetype: telemetry
    messages:
      message:
        payload:
          type: object

  setpointCommand:
    address: thermo2mqtt/setpoint/set
    x-cosalette-archetype: command
    messages:
      message:
        payload:
          type: object

operations:
  publishTemperatureState:
    action: send
    channel:
      $ref: '#/channels/temperatureState'
  receiveSetpointCommand:
    action: receive
    channel:
      $ref: '#/channels/setpointCommand'

The scaffolded payloads are type: object — add properties and constraints by hand.

2 — Add payload constraints¶

Edit the generated YAML to specify required fields and types:

channels:
  temperatureState:
    address: thermo2mqtt/temperature/state
    x-cosalette-app: thermo2mqtt
    x-cosalette-archetype: telemetry
    messages:
      reading:
        payload:
          type: object
          required: [temperature, unit]
          properties:
            temperature:
              type: number
              x-cosalette-consumer:
                device_class: temperature
                unit: "°C"
                display_name: Room Temperature
                state_class: measurement
            unit:
              type: string
              enum: [celsius, fahrenheit]

The x-cosalette-consumer annotation carries metadata for downstream consumer code generation (e.g. home automation integrations, dashboard provisioning). It is optional — omit it if you do not need consumer code generation.

3 — Validate the schema document¶

cosalette schema validate schema.yaml

✅ Schema validated: thermo2mqtt v0.1.0
   AsyncAPI version: 3.0.0
   Channels: 2
   Schema type: single-app

This checks structure and cosalette-specific extension syntax. It does not run the app — it validates the schema file alone.

4 — Check app registrations against the schema¶

cosalette schema check --app thermo2mqtt.app:app --schema schema.yaml

Schema: schema.yaml (v0.1.0)
App:    thermo2mqtt

✓ temperature — OK
✓ setpoint — OK

Result: 0 violations, 2 compliant
Exit code: 0

If a device is missing or a scope rule is violated, check exits with code 1 and prints the violation:

✗ setpoint — MISSING
    Schema expects device 'setpoint' but no registration found

Result: 1 violations, 1 compliant
Exit code: 1

5 — Enable enforcement in the app¶

Schema enforcement is configured through the framework's nested settings model under the schema key. The two relevant fields are:

Set them in your .env file (or environment):

SCHEMA__PATH=/etc/cosalette/thermo2mqtt-schema.yaml
SCHEMA__ENFORCEMENT=warn

Or in a settings subclass if you prefer code-level defaults:

from cosalette import App, Settings
from cosalette._settings import SchemaSettings

class MySettings(Settings):
    schema_: SchemaSettings = SchemaSettings(
        path="/etc/cosalette/thermo2mqtt-schema.yaml",
        enforcement="warn",
    )

app = App(name="thermo2mqtt", settings_class=MySettings)

Enforcement Modes¶

Configure in the schema file:

x-cosalette-enforcement:
  mode: warn          # off | warn | strict
  on_configure: true  # validate device registrations at startup
  on_publish: false   # validate payload shape at publish time (dev only)
  network_level: false

Network-Level Schema¶

For a fleet of multiple cosalette apps, a network-level schema defines the entire MQTT topology in one file. Each app validates against its own slice.

When to use it¶

A network schema is the primary use case for cross-app validation:

• One app renames a topic → the network schema flags it; the deploy is blocked before the change reaches the broker.
• An app adds a new channel → cosalette schema check reports it as "extra" (not a violation) so you can decide whether to promote it to the schema.
• A CI/CD gate validates all apps in a single step before any of them are deployed.

Network schema structure¶

asyncapi: 3.0.0
info:
  title: My MQTT Network
  version: 1.0.0

x-cosalette-enforcement:
  mode: warn
  network_level: true   # marks this as a network-level schema

channels:
  thermoTemperatureState:
    address: thermo2mqtt/temperature/state
    x-cosalette-app: thermo2mqtt           # channel belongs to this app
    x-cosalette-archetype: telemetry
    messages:
      reading:
        payload:
          type: object
          required: [temperature]
          properties:
            temperature:
              type: number

  airsenseAirQualityState:
    address: airsense2mqtt/airquality/state
    x-cosalette-app: airsense2mqtt
    x-cosalette-archetype: telemetry
    messages:
      reading:
        payload:
          type: object
          required: [co2, humidity]
          properties:
            co2:
              type: integer
            humidity:
              type: number

Extract an app's slice¶

The slice command filters a network schema to a single app's channels:

cosalette schema slice --network /etc/cosalette/network-schema.yaml --app thermo2mqtt

asyncapi: 3.0.0
info:
  title: thermo2mqtt
  version: 2.0.0

x-cosalette-enforcement:
  mode: warn
  on_configure: true
  on_publish: false
  network_level: false

channels:
  thermoTemperatureState:
    address: thermo2mqtt/temperature/state
    x-cosalette-app: thermo2mqtt
    x-cosalette-archetype: telemetry
    ...

You can pipe this directly to a file or use it for per-app validation:

cosalette schema slice \
  --network /etc/cosalette/network-schema.yaml \
  --app thermo2mqtt > /etc/cosalette/thermo2mqtt-schema.yaml

Check against a network schema¶

check automatically filters the network schema to the app's slice:

cosalette schema check \
  --app thermo2mqtt.app:app \
  --schema /etc/cosalette/network-schema.yaml

The command detects network_level: true, extracts the thermo2mqtt slice, and validates the app's registrations against it. No separate slice step needed.

Reference schema¶

See docs/assets/reference-network-schema.yaml for a complete annotated example covering a three-app smart-home fleet (thermo2mqtt, airsense2mqtt, solarray2mqtt) with telemetry, command, and fleet-wide availability channels.

Deployment Integration¶

cosalette schema check is a standard subprocess that exits 0 on compliance and 1 on violations, so it integrates cleanly into any deploy toolchain.

Shell / CI scripts¶

The simplest form — run before starting each app:

cosalette schema check \
  --app myapp.app:app \
  --schema /etc/cosalette/network-schema.yaml || exit 1

# Start the app only if validation passed
myapp start

This works in any environment: bare-metal init scripts, Docker entrypoints, GitHub Actions steps, GitLab CI jobs, or Makefile targets.

GitHub Actions example¶

- name: Validate schema
  run: |
    cosalette schema check \
      --app myapp.app:app \
      --schema schemas/network-schema.yaml

Ansible example¶

For teams using Ansible to manage hosts, the pattern below deploys the schema file first and validates before starting the service:

# tasks/deploy-myapp.yml

- name: Deploy network schema
  ansible.builtin.copy:
    src: files/network-schema.yaml
    dest: /etc/cosalette/network-schema.yaml
    mode: "0644"

- name: Validate myapp against network schema
  ansible.builtin.command:
    cmd: >
      cosalette schema check
        --app myapp.app:app
        --schema /etc/cosalette/network-schema.yaml
  changed_when: false
  failed_when: result.rc != 0
  register: result

- name: Start myapp service
  ansible.builtin.systemd:
    name: myapp
    state: started
  when: result.rc == 0

The exit-code contract is the same regardless of toolchain — check failing blocks the next step.

x-cosalette Extension Reference¶

Channel-level extensions¶

Property-level extensions¶

Document-level enforcement config¶

x-cosalette-enforcement:
  mode: warn          # off | warn | strict
  on_configure: true  # validate at startup
  on_publish: false   # validate payload at publish time
  network_level: false

CLI Reference¶

Consumer Code Generation¶

Properties annotated with x-cosalette-consumer can be transformed into consumer platform configurations automatically. This eliminates hand-maintaining discovery payloads and configuration files — the AsyncAPI schema becomes the single source of truth.

Home Assistant MQTT Discovery¶

Generate HA discovery payloads that Home Assistant accepts via its MQTT discovery protocol:

cosalette schema ha-discovery network.yaml

Output is a JSON array of {topic, config} objects — one per annotated property. Each object contains the discovery topic and the full config payload. Publish these as retained messages and HA will auto-create entities.

Component inference: When x-cosalette-ha-discovery.component is not set, the component is inferred from archetype and JSON schema type:

Example:

# In your AsyncAPI schema
temperature:
  type: number
  x-cosalette-consumer:
    device_class: temperature
    unit: '°C'
    display_name: 'Heating Water Temperature'
    state_class: measurement
  x-cosalette-ha-discovery:
    expire_after: 300

Produces a discovery payload at homeassistant/sensor/<app>/<device>_temperature/config with device_class, unit_of_measurement, state_class, value_template, and expire_after fields set correctly.

OpenHAB Configuration¶

Generate OpenHAB .things and .items files:

cosalette schema openhab network.yaml --output both
cosalette schema openhab network.yaml --output things
cosalette schema openhab network.yaml --output items

Things use JSONPATH transformations to extract individual properties from JSON payloads. Items are typed according to device_class (e.g. Number:Temperature) or explicit x-cosalette-openhab.item_type overrides.

Further Reading¶

• ADR-033 — MQTT Schema Enforcement — decision record: format choice, distribution model, enforcement modes.
• Reference Network Schema — annotated three-app fleet schema.