Skip to content

FR: use dataclasses instead of Struct #40

@rsyring

Description

@rsyring

The Struct object leaves a lot to be desired. Everything beyond the first level of access is still a dict, you can't print it, and there is no help for developers inspecting the code to know what the data structure being returned is.

I propose using dataclasses for this:

from __future__ import annotations

from dataclasses import asdict, dataclass
from datetime import datetime
import json
from typing import Any

@dataclass
class Attributes:
    group_name: str | None
    key: str | None
    code: str

    @classmethod
    def from_dict(cls, data: dict) -> Attributes:
        return cls(
            group_name=data.get('group_name'),
            key=data.get('key'),
            code=data['code'],
        )


@dataclass
class LatestEvent:
    stamp: float
    msg: str | None
    event: str
    metrics: dict
    client: str
    host: str | None
    ip: str

    @classmethod
    def from_dict(cls, data: dict) -> LatestEvent:
        return cls(
            stamp=data['stamp'],
            msg=data.get('msg'),
            event=data['event'],
            metrics=data.get('metrics', {}),
            client=data['client'],
            host=data.get('host'),
            ip=data['ip'],
        )


@dataclass
class LatestIssue:
    stamp: float
    state: str

    @classmethod
    def from_dict(cls, data: dict) -> LatestIssue:
        return cls(
            stamp=data['stamp'],
            state=data['state'],
        )


@dataclass
class Monitor:
    attributes: Attributes
    assertions: list[str]
    created: datetime
    disabled: bool
    failure_tolerance: int | None
    grace_seconds: int
    consecutive_alert_threshold: int
    group: str | None
    initialized: bool
    key: str
    latest_event: LatestEvent | None
    latest_events: list[LatestEvent] | None
    latest_issue: LatestIssue | None
    latest_invocations: list[Any] | None
    public_badge_url: str
    metadata: dict | None
    name: str
    next_expected_at: datetime | None
    note: str | None
    notify: list[str]
    passing: bool
    paused: bool
    platform: str
    realert_interval: str
    request: str | None
    running: bool
    schedule: str | None
    schedule_tolerance: int
    tags: list[str]
    timezone: str
    type: str
    environments: list[str]
    statuspages: list[Any]
    site: str | None

    @classmethod
    def from_dict(cls, data: dict) -> Monitor:
        return cls(
            attributes=Attributes.from_dict(data['attributes']),
            assertions=data.get('assertions', []),
            created=datetime.fromisoformat(data['created']),
            disabled=data['disabled'],
            failure_tolerance=data.get('failure_tolerance'),
            grace_seconds=data['grace_seconds'],
            consecutive_alert_threshold=data['consecutive_alert_threshold'],
            group=data.get('group'),
            initialized=data['initialized'],
            key=data['key'],
            latest_event=LatestEvent.from_dict(data['latest_event'])
            if data['latest_event']
            else None,
            latest_events=(
                [LatestEvent.from_dict(e) for e in data['latest_events']]
                if data.get('latest_events') is not None
                else None
            ),
            latest_issue=(
                LatestIssue.from_dict(data['latest_issue']) if data.get('latest_issue') else None
            ),
            latest_invocations=data.get('latest_invocations'),
            public_badge_url=data['public_badge_url'],
            metadata=data.get('metadata'),
            name=data['name'],
            next_expected_at=(
                datetime.fromisoformat(data['next_expected_at'])
                if data.get('next_expected_at')
                else None
            ),
            note=data.get('note'),
            notify=data.get('notify', []),
            passing=data['passing'],
            paused=data['paused'],
            platform=data['platform'],
            realert_interval=data['realert_interval'],
            request=data.get('request'),
            running=data['running'],
            schedule=data.get('schedule'),
            schedule_tolerance=data['schedule_tolerance'],
            tags=data.get('tags', []),
            timezone=data['timezone'],
            type=data['type'],
            environments=data.get('environments', []),
            statuspages=data.get('statuspages', []),
            site=data.get('site'),
        )

    def __str__(self):
        return json.dumps(asdict(self), indent=2, sort_keys=True, cls=DateTimeEncoder)


class DateTimeEncoder(json.JSONEncoder):
    def default(self, o):
        if isinstance(o, datetime):
            return o.isoformat()
        return super().default(o)

It will need some tweaking to match your actual datastructures but this is working for me with the monitors I'm currently configuring:

    monitors = cronitor.Monitor.put(...)
    for m in monitors:
        m = Monitor.from_dict(m.data.__dict__)
        print(m)

Gives for a single monitor:

{
  "assertions": [
    "response.body contains moleliminator ok",
    "response.code = 200",
    "response.time <= 2s"
  ],
  "attributes": {
    "code": "Ydl0ol",
    "group_name": null,
    "key": "mole-up-prod"
  },
  "consecutive_alert_threshold": 10,
  "created": "2025-06-28T21:13:28+00:00",
  "disabled": false,
  "environments": [
    "production"
  ],
  "failure_tolerance": 3,
  "grace_seconds": 0,
  "group": null,
  "initialized": null,
  "key": "mole-up-prod",
  "latest_event": null,
  "latest_events": null,
  "latest_invocations": null,
  "latest_issue": null,
  "metadata": null,
  "name": "Moleliminator: Up (prod)",
  "next_expected_at": null,
  "note": null,
  "notify": [
    "dev-team-b"
  ],
  "passing": null,
  "paused": false,
  "platform": null,
  "public_badge_url": null,
  "realert_interval": "every 30 minutes",
  "request": {
    "body": null,
    "cookies": {},
    "follow_redirects": true,
    "headers": {},
    "method": "GET",
    "regions": [
      "us-west-1",
      "us-east-2"
    ],
    "timeout_seconds": 15,
    "url": "https://level12.io/health-check",
    "verify_ssl": true
  },
  "running": null,
  "schedule": "every 1 minute",
  "schedule_tolerance": null,
  "site": null,
  "statuspages": [],
  "tags": [],
  "timezone": null,
  "type": "check"
}

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type
    No fields configured for issues without a type.

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions