Client Class Reference

from pydoover.cloud.api import Client

client = Client(
    username: str = None,
    password: str = None,
    token: str = None,
    token_expires: datetime = None,
    base_url: str = "https://my.doover.dev",
    agent_id: str = None,
    verify: bool = True,
    login_callback: Callable = None,
    config_profile: str = "default",
    debug: bool = False,
)

client.login()

token, expires_at, agent_id = client.fetch_token()

agent = client.get_agent(agent_id: str) -> Agent

agent = client.get_agent("9fb5d629-ce7f-4b08-b17a-c267cbcd0427")
print(f"Agent name: {agent.name}")
print(f"Agent type: {agent.type}")

agents = client.get_agent_list() -> list[Agent]

agents = client.get_agent_list()
for agent in agents:
    print(f"{agent.name}: {agent.key}")

channel = client.get_channel(channel_id: str) -> Channel | Processor | Task

channel = client.get_channel("d1c7e8e3-f47b-4c68-86d7-65054d9e97d3")
print(f"Channel: {channel.name}")
print(f"Aggregate: {channel.aggregate}")

channel = client.get_channel_named(
    channel_name: str,
    agent_id: str
) -> Channel | Processor | Task

channel = client.get_channel_named("status", "9fb5d629-ce7f-4b08-b17a-c267cbcd0427")

channel = client.create_channel(
    channel_name: str,
    agent_id: str
) -> Channel | Processor | Task

# Create a regular channel
channel = client.create_channel("sensor-data", agent.key)

# Create a processor channel
processor = client.create_channel("#my-processor", agent.key)

# Create a task channel (requires existing processor)
task = client.create_task("my-task", agent.key, processor.key)

processor = client.create_processor(
    processor_name: str,
    agent_id: str
) -> Processor

task = client.create_task(
    task_name: str,
    agent_id: str,
    processor_id: str
) -> Task

client.publish_to_channel(
    channel_id: str,
    data: Any,
    save_log: bool = True,
    log_aggregate: bool = False,
    override_aggregate: bool = False,
    timestamp: datetime = None
)

client.publish_to_channel(
    channel_id="d1c7e8e3-f47b-4c68-86d7-65054d9e97d3",
    data={"temperature": 25.5, "humidity": 60},
    save_log=True
)

client.publish_to_channel_name(
    agent_id: str,
    channel_name: str,
    data: Any,
    save_log: bool = True,
    log_aggregate: bool = False,
    override_aggregate: bool = False,
    timestamp: datetime = None
)

messages = client.get_channel_messages(
    channel_id: str,
    num_messages: int = None
) -> list[Message]

messages = client.get_channel_messages("channel-uuid", num_messages=10)
for msg in messages:
    print(f"[{msg.timestamp}] {msg.fetch_payload()}")

messages = client.get_channel_messages_in_window(
    channel_id: str,
    start: datetime,
    end: datetime
) -> list[Message]

from datetime import datetime, timedelta

end = datetime.now()
start = end - timedelta(hours=24)

messages = client.get_channel_messages_in_window("channel-uuid", start, end)
print(f"Found {len(messages)} messages in the last 24 hours")

message = client.get_message(
    channel_id: str,
    message_id: str
) -> Message

result = client.subscribe_to_channel(
    channel_id: str,
    task_id: str
) -> bool

result = client.unsubscribe_from_channel(
    channel_id: str,
    task_id: str
) -> bool

tunnels = client.get_tunnels(
    agent_id: str,
    show_choices: bool = False
)

tunnel = client.get_tunnel(tunnel_id: str)

tunnel = client.create_tunnel(agent_id: str, **data)

client.activate_tunnel(tunnel_id: str)

client.deactivate_tunnel(tunnel_id: str)

client.delete_tunnel(tunnel_id: str)

client.patch_tunnel(tunnel_id: str, **data)

apps = client.get_applications()

app = client.get_application(key: str) -> Application

key = client.create_application(
    application: Application,
    is_staging: bool = False
) -> str

client.update_application(
    application: Application,
    is_staging: bool = False
)

from pydoover.cloud.api import Client, NotFound
from datetime import datetime, timedelta

# Initialize client
client = Client(config_profile="production")

# List all agents
agents = client.get_agent_list()
print(f"Found {len(agents)} agents")

# Work with a specific agent
agent = agents[0]
print(f"Working with agent: {agent.name}")

# Create or get a channel
channel = client.create_channel("sensor-readings", agent.key)

# Publish some data
client.publish_to_channel(
    channel.key,
    data={
        "temperature": 23.5,
        "humidity": 45,
        "timestamp": datetime.now().isoformat()
    }
)

# Read back recent messages
messages = client.get_channel_messages(channel.key, num_messages=5)
for msg in messages:
    print(f"Message at {msg.timestamp}: {msg.fetch_payload()}")

# Get messages from the last hour
end = datetime.now()
start = end - timedelta(hours=1)
recent = client.get_channel_messages_in_window(channel.key, start, end)
print(f"Found {len(recent)} messages in the last hour")