Skip to content

Channels

Channels are the fundamental data containers in the Doover platform. Each channel belongs to an agent and stores a stream of messages along with an aggregate (a single document representing the channel's current state). The DataClient provides methods for listing, fetching, creating, and updating channels.

Aggregates vs Messages

Before diving into the API, it is important to understand the two types of data a channel holds:

  • Messages are an ordered, append-only stream of data points. Each message has a snowflake ID that encodes its timestamp. Messages are used for historical data, timeseries, and event logs.
  • Aggregates are a single mutable document per channel that represents the channel's current state. Aggregates can be patched (merged) or replaced entirely. They are ideal for configuration, status, or any data where only the latest value matters.

When you fetch a channel, you can optionally include its aggregate and daily summaries in the response.

Listing Channels

Retrieve all channels for a given agent. Each returned Channel object optionally includes the channel's aggregate data.

from pydoover.api import DataClient

client = DataClient(profile="default")

# List all channels with aggregates included
channels = client.list_channels(
    agent_id=12345,
    include_aggregate=True,
    include_daily_summaries=True,
)

for ch in channels:
    print(f"{ch.name} (id={ch.id})")

The include_aggregate and include_daily_summaries parameters both default to True. Set them to False to reduce response size when you only need channel metadata.

Fetching a Single Channel

Retrieve a specific channel by name:

# Fetch a single channel with its aggregate
channel = client.fetch_channel(
    agent_id=12345,
    channel_name="telemetry",
    include_aggregate=True,
)

print(f"Channel: {channel.name}")
print(f"ID: {channel.id}")

Creating a Channel

Create a new channel on an agent. Returns the new channel's snowflake ID.

# Create a basic public channel
channel_id = client.create_channel(
    agent_id=12345,
    channel_name="sensor_data",
)
print(f"Created channel with ID: {channel_id}")

You can also specify schemas and privacy settings. Schemas define the expected structure of messages and aggregates for the channel:

# Create a private channel with schemas
channel_id = client.create_channel(
    agent_id=12345,
    channel_name="internal_config",
    is_private=True,
    message_schema={
        "type": "object",
        "properties": {
            "setting": {"type": "string"},
            "value": {"type": "number"},
        },
    },
    aggregate_schema={
        "type": "object",
        "properties": {
            "version": {"type": "integer"},
        },
    },
)

Updating a Channel (PUT)

Replace a channel's configuration entirely using put_channel. This creates the channel if it does not exist, or replaces it if it does. Unlike create_channel, this returns the full Channel object:

# Create or replace a channel
channel = client.put_channel(
    agent_id=12345,
    channel_name="sensor_data",
    is_private=False,
    message_schema={"type": "object"},
    aggregate_schema={"type": "object"},
)

print(f"Channel updated: {channel.name} (id={channel.id})")
Warning

put_channel performs a full replacement. Any fields you omit (like message_schema) will be cleared on the channel. Use it when you want to ensure the channel matches an exact specification.

Working with Aggregates

The aggregate is the mutable "current state" document on a channel. You can fetch and update it separately from messages.

Fetching an Aggregate

# Fetch just the aggregate for a channel
aggregate = client.fetch_channel_aggregate(
    agent_id=12345,
    channel_name="device_status",
)

print(aggregate.data)

Updating an Aggregate

By default, update_channel_aggregate performs a merge (PATCH). Only the keys you provide are updated:

# Merge new data into the aggregate
updated = client.update_channel_aggregate(
    agent_id=12345,
    channel_name="device_status",
    data={"online": True, "last_seen": "2026-05-07T10:00:00Z"},
)

To replace the aggregate entirely, set replace_data=True:

# Replace the entire aggregate
client.update_channel_aggregate(
    agent_id=12345,
    channel_name="device_status",
    data={"online": False},
    replace_data=True,
)

You can also suppress the response body for faster writes when you do not need the result:

# Fire-and-forget update (no response body returned)
client.update_channel_aggregate(
    agent_id=12345,
    channel_name="device_status",
    data={"heartbeat": 1234567890},
    suppress_response=True,
)

Method Reference

MethodReturnsDescription
list_channels(agent_id, ...)list[Channel]List all channels for an agent
fetch_channel(agent_id, channel_name, ...)ChannelFetch a single channel by name
create_channel(agent_id, channel_name, ...)intCreate a channel, returns snowflake ID
put_channel(agent_id, channel_name, ...)ChannelCreate or replace a channel
fetch_channel_aggregate(agent_id, channel_name)AggregateFetch the channel's aggregate
update_channel_aggregate(agent_id, channel_name, data, ...)Aggregate | NoneUpdate the channel's aggregate

Related Pages

  • Messages -- reading and writing messages on channels
  • Data Client -- DataClient constructor and configuration
  • Alarms -- setting up alarm rules on channels