Heartbeat¶

The Heartbeat module provides functionality to manage and interact with heartbeat monitors.

Features¶

Create and manage heartbeat monitors
Send heartbeat events
Configure alert settings
Manage maintenance windows

Examples¶

Basic Usage¶

Create and use a heartbeat monitor

import upassist

# Configure API key
upassist.config.API_KEY = 'your-api-key'

# Create a heartbeat instance
heartbeat = upassist.Heartbeat("my-heartbeat")

# Send a heartbeat event
heartbeat.event()

Advanced Configuration¶

Create a heartbeat with custom settings

from upassist.entities.heartbeat import Heartbeat, HeartbeatCreateSchema
from datetime import time

heartbeat = Heartbeat()
new_heartbeat = heartbeat.create(
    HeartbeatCreateSchema(
        name="Production Server Monitor",
        description="Monitors the production server health",
        group_id="prod-servers",
        slug="prod-server-monitor",
        fetch_interval=300,  # Check every 5 minutes
        confirmation_period=60,  # Wait 1 minute before alerting
        realert_period=3600,  # Re-alert every hour
        alerts_on=True,
        maintenance_window_from=time(2, 0),  # 02:00
        maintenance_window_until=time(3, 0),  # 03:00
        maintenance_window_timezone="UTC",
        alert_week_days=[1, 2, 3, 4, 5]  # Monday to Friday
    )
)

Managing Heartbeats¶

List, pause, and manage heartbeats

# List all heartbeats
heartbeats = heartbeat.list()

# Get details of a specific heartbeat
details = heartbeat.detail()

# Pause a heartbeat
heartbeat.pause()

# Resume a heartbeat
heartbeat.unpause()

# Delete a heartbeat
heartbeat.delete()

API Reference¶

`Heartbeat` ¶

Bases: BaseEntity

A class representing a heartbeat monitoring entity.

This class provides functionality to manage and interact with heartbeat monitors, including creating, listing, pausing, and managing heartbeat events.

Source code in upassist/entities/heartbeat/heartbeat.py

class Heartbeat(BaseEntity):
    """A class representing a heartbeat monitoring entity.

    This class provides functionality to manage and interact with heartbeat monitors,
    including creating, listing, pausing, and managing heartbeat events.
    """

    base_heartbeat_event_api_url: str = "https://heartbeats.upassist.cloud/api"

    def __init__(
        self,
        heartbeat_slug: str | None = None,
        api_key: str | None = None,
        api_version: str | None = None,
        api_client_cls: type[AbstractAPIClient] = SyncAPIClient,
    ):
        """Initialize a new Heartbeat instance.

        Args:
            heartbeat_slug: Unique identifier for the heartbeat
            api_key: API key for authentication
            api_version: API version to use
            api_client_cls: Class to use for API client implementation
        """
        super().__init__(
            api_key=api_key or config.API_KEY,
            api_version=api_version or config.API_VERSION,
            api_client_cls=api_client_cls,
        )
        self.heartbeat_slug = heartbeat_slug

    def list(
        self, q: str | None = None, page: int | None = None, per_page: int | None = None
    ) -> HeartbeatPaginatedSchema:
        """List all heartbeats with optional filtering and pagination.

        Args:
            q: Search query string
            page: Page number for pagination
            per_page: Number of items per page

        Returns:
            HeartbeatPaginatedSchema containing the list of heartbeats
        """
        response = self.api_client.get(
            self.base_heartbeats_api_url,
            params={
                "q": q,
                "page": page,
                "per_page": per_page,
            },
        )
        return HeartbeatPaginatedSchema.model_validate(response)

    @property
    def base_heartbeats_api_url(self) -> str:
        """Get the base URL for heartbeat API endpoints.

        Returns:
            Base URL for heartbeat API endpoints
        """
        return f"{self.api_client.base_api_url}/heartbeats"

    @heartbeat_slug_required
    def detail(self) -> HeartbeatDetailSchema:
        """Get detailed information about a specific heartbeat.

        Returns:
            HeartbeatDetailSchema containing detailed heartbeat information
        """
        response = self.api_client.get(f"{self.base_heartbeats_api_url}/{self.heartbeat_slug}")
        return HeartbeatDetailSchema.model_validate(response)

    @heartbeat_slug_required
    def pause(self):
        """Pause the heartbeat monitoring."""
        return self.api_client.patch(
            f"{self.base_heartbeats_api_url}/{self.heartbeat_slug}/pause",
        )

    @heartbeat_slug_required
    def unpause(self):
        """Resume the heartbeat monitoring."""
        return self.api_client.patch(
            f"{self.base_heartbeats_api_url}/{self.heartbeat_slug}/unpause",
        )

    @heartbeat_slug_required
    def delete(self) -> None:
        """Delete the heartbeat."""
        self.api_client.delete(f"{self.base_heartbeats_api_url}/{self.heartbeat_slug}")

    @heartbeat_slug_required
    def event(self) -> DetailResponse:
        """Get the current event status of the heartbeat.

        Returns:
            DetailResponse containing the event status
        """
        return self.api_client.get(f"{self.base_heartbeat_event_api_url}/event/{self.heartbeat_slug}")

    def create(self, heartbeat: HeartbeatCreateSchema) -> HeartbeatSchema:
        """Create a new heartbeat.

        Args:
            heartbeat: The heartbeat configuration.

        Returns:
            HeartbeatSchema containing the created heartbeat information
        """
        response = self.api_client.post(
            self.base_heartbeats_api_url,
            json=heartbeat.model_dump(exclude_unset=True),
        )
        return HeartbeatSchema.model_validate(response)

`base_heartbeats_api_url` `property` ¶

Get the base URL for heartbeat API endpoints.

Returns:

Type	Description
`str`	Base URL for heartbeat API endpoints

`init(heartbeat_slug=None, api_key=None, api_version=None, api_client_cls=SyncAPIClient)` ¶

Initialize a new Heartbeat instance.

Parameters:

Name	Type	Description	Default
`heartbeat_slug`	`str \| None`	Unique identifier for the heartbeat	`None`
`api_key`	`str \| None`	API key for authentication	`None`
`api_version`	`str \| None`	API version to use	`None`
`api_client_cls`	`type[AbstractAPIClient]`	Class to use for API client implementation	`SyncAPIClient`

Source code in upassist/entities/heartbeat/heartbeat.py

def __init__(
    self,
    heartbeat_slug: str | None = None,
    api_key: str | None = None,
    api_version: str | None = None,
    api_client_cls: type[AbstractAPIClient] = SyncAPIClient,
):
    """Initialize a new Heartbeat instance.

    Args:
        heartbeat_slug: Unique identifier for the heartbeat
        api_key: API key for authentication
        api_version: API version to use
        api_client_cls: Class to use for API client implementation
    """
    super().__init__(
        api_key=api_key or config.API_KEY,
        api_version=api_version or config.API_VERSION,
        api_client_cls=api_client_cls,
    )
    self.heartbeat_slug = heartbeat_slug

`create(heartbeat)` ¶

Create a new heartbeat.

Parameters:

Name	Type	Description	Default
`heartbeat`	`HeartbeatCreateSchema`	The heartbeat configuration.	required

Returns:

Type	Description
`HeartbeatSchema`	HeartbeatSchema containing the created heartbeat information

Source code in upassist/entities/heartbeat/heartbeat.py

def create(self, heartbeat: HeartbeatCreateSchema) -> HeartbeatSchema:
    """Create a new heartbeat.

    Args:
        heartbeat: The heartbeat configuration.

    Returns:
        HeartbeatSchema containing the created heartbeat information
    """
    response = self.api_client.post(
        self.base_heartbeats_api_url,
        json=heartbeat.model_dump(exclude_unset=True),
    )
    return HeartbeatSchema.model_validate(response)

`delete()` ¶

Delete the heartbeat.

Source code in upassist/entities/heartbeat/heartbeat.py

@heartbeat_slug_required
def delete(self) -> None:
    """Delete the heartbeat."""
    self.api_client.delete(f"{self.base_heartbeats_api_url}/{self.heartbeat_slug}")

`detail()` ¶

Get detailed information about a specific heartbeat.

Returns:

Type	Description
`HeartbeatDetailSchema`	HeartbeatDetailSchema containing detailed heartbeat information

Source code in upassist/entities/heartbeat/heartbeat.py

@heartbeat_slug_required
def detail(self) -> HeartbeatDetailSchema:
    """Get detailed information about a specific heartbeat.

    Returns:
        HeartbeatDetailSchema containing detailed heartbeat information
    """
    response = self.api_client.get(f"{self.base_heartbeats_api_url}/{self.heartbeat_slug}")
    return HeartbeatDetailSchema.model_validate(response)

`event()` ¶

Get the current event status of the heartbeat.

Returns:

Type	Description
`DetailResponse`	DetailResponse containing the event status

Source code in upassist/entities/heartbeat/heartbeat.py

@heartbeat_slug_required
def event(self) -> DetailResponse:
    """Get the current event status of the heartbeat.

    Returns:
        DetailResponse containing the event status
    """
    return self.api_client.get(f"{self.base_heartbeat_event_api_url}/event/{self.heartbeat_slug}")

`list(q=None, page=None, per_page=None)` ¶

List all heartbeats with optional filtering and pagination.

Parameters:

Name	Type	Description	Default
`q`	`str \| None`	Search query string	`None`
`page`	`int \| None`	Page number for pagination	`None`
`per_page`	`int \| None`	Number of items per page	`None`

Returns:

Type	Description
`HeartbeatPaginatedSchema`	HeartbeatPaginatedSchema containing the list of heartbeats

Source code in upassist/entities/heartbeat/heartbeat.py

def list(
    self, q: str | None = None, page: int | None = None, per_page: int | None = None
) -> HeartbeatPaginatedSchema:
    """List all heartbeats with optional filtering and pagination.

    Args:
        q: Search query string
        page: Page number for pagination
        per_page: Number of items per page

    Returns:
        HeartbeatPaginatedSchema containing the list of heartbeats
    """
    response = self.api_client.get(
        self.base_heartbeats_api_url,
        params={
            "q": q,
            "page": page,
            "per_page": per_page,
        },
    )
    return HeartbeatPaginatedSchema.model_validate(response)

`pause()` ¶

Pause the heartbeat monitoring.

Source code in upassist/entities/heartbeat/heartbeat.py

@heartbeat_slug_required
def pause(self):
    """Pause the heartbeat monitoring."""
    return self.api_client.patch(
        f"{self.base_heartbeats_api_url}/{self.heartbeat_slug}/pause",
    )

`unpause()` ¶

Resume the heartbeat monitoring.

Source code in upassist/entities/heartbeat/heartbeat.py

@heartbeat_slug_required
def unpause(self):
    """Resume the heartbeat monitoring."""
    return self.api_client.patch(
        f"{self.base_heartbeats_api_url}/{self.heartbeat_slug}/unpause",
    )

`HeartbeatCreateSchema` ¶

Bases: BaseModel

Schema for creating a new heartbeat.

Attributes:

Name	Type	Description
`name`	`str`	Name of the heartbeat
`description`	`str \| None`	Optional description of the heartbeat
`group_id`	`UUID \| None`	Optional UUID of the group this heartbeat belongs to
`slug`	`str \| None`	Optional custom slug for the heartbeat
`fetch_interval`	`PositiveInt`	Interval in seconds between heartbeat checks (60-100000)
`confirmation_period`	`NonNegativeInt`	Time in seconds to wait before confirming an incident (0-100000)
`realert_period`	`int \| None`	Optional time in seconds before re-alerting about an incident
`alerts_on`	`bool`	Whether to send alerts on incidents
`paused`	`bool`	Whether the heartbeat is paused
`meta`	`dict \| None`	Optional metadata dictionary
`call`	`bool`	Whether to make phone calls for alerts
`send_sms`	`bool`	Whether to send SMS alerts
`send_email`	`bool`	Whether to send email alerts
`send_push_notification`	`bool`	Whether to send push notifications
`maintenance_window_from`	`time \| None`	Optional start time for maintenance window
`maintenance_window_until`	`time \| None`	Optional end time for maintenance window
`maintenance_window_timezone`	`TimeZoneName`	Timezone for maintenance window times
`alert_week_days`	`list[int] \| None`	List of days (0-6) when alerts should be sent

Source code in upassist/entities/heartbeat/schemas.py

class HeartbeatCreateSchema(BaseModel):
    """Schema for creating a new heartbeat.

    Attributes:
        name: Name of the heartbeat
        description: Optional description of the heartbeat
        group_id: Optional UUID of the group this heartbeat belongs to
        slug: Optional custom slug for the heartbeat
        fetch_interval: Interval in seconds between heartbeat checks (60-100000)
        confirmation_period: Time in seconds to wait before confirming an incident (0-100000)
        realert_period: Optional time in seconds before re-alerting about an incident
        alerts_on: Whether to send alerts on incidents
        paused: Whether the heartbeat is paused
        meta: Optional metadata dictionary
        call: Whether to make phone calls for alerts
        send_sms: Whether to send SMS alerts
        send_email: Whether to send email alerts
        send_push_notification: Whether to send push notifications
        maintenance_window_from: Optional start time for maintenance window
        maintenance_window_until: Optional end time for maintenance window
        maintenance_window_timezone: Timezone for maintenance window times
        alert_week_days: List of days (0-6) when alerts should be sent
    """

    name: str
    description: str | None = None
    group_id: UUID | None = None
    slug: str | None = None
    fetch_interval: PositiveInt = Field(default=180, ge=60, le=10**5)
    confirmation_period: NonNegativeInt = Field(default=0, le=10**5)
    realert_period: int | None = Field(default=None)
    alerts_on: bool = True
    paused: bool = False
    meta: dict | None = None
    call: bool = False
    send_sms: bool = False
    send_email: bool = True
    send_push_notification: bool = False
    maintenance_window_from: time | None = None
    maintenance_window_until: time | None = None
    maintenance_window_timezone: TimeZoneName = Field(default=TimeZoneName("Europe/Belfast"))
    alert_week_days: list[int] | None = Field(default_factory=lambda: [0, 1, 2, 3, 4, 5, 6])

`HeartbeatDetailSchema` ¶

Bases: HeartbeatSchema

Schema for detailed heartbeat information.

Additional attributes beyond HeartbeatSchema

Source code in upassist/entities/heartbeat/schemas.py

class HeartbeatDetailSchema(HeartbeatSchema):
    """Schema for detailed heartbeat information.

    Additional attributes beyond HeartbeatSchema:
        description: Optional description of the heartbeat
        incidents_count: Number of incidents
        opened_incident_id: UUID of currently open incident if any
        confirmation_period: Time to wait before confirming incidents
        realert_period: Time before re-alerting about unresolved incidents
        meta: Optional metadata dictionary
        incident_stats: List of incident statistics
    """

    description: str | None = None
    incidents_count: int | None = Field(default=0)
    opened_incident_id: UUID | None = None
    confirmation_period: NonNegativeInt | None = Field(
        default=None,
        description="Time in seconds to wait until beat before starting images",
    )
    realert_period: NonNegativeInt | None = Field(
        default=None,
        description="Time in seconds to notify users again abouts unresolved incident",
    )
    meta: dict | None = None
    incident_stats: list[IncidentStatsSchema]

`HeartbeatListSchema` ¶

Bases: HeartbeatSchema

Schema for heartbeat list items.

Additional attributes beyond HeartbeatSchema

Source code in upassist/entities/heartbeat/schemas.py

class HeartbeatListSchema(HeartbeatSchema):
    """Schema for heartbeat list items.

    Additional attributes beyond HeartbeatSchema:
        incidents_count: Number of incidents
        opened_incident_id: UUID of currently open incident if any
    """

    incidents_count: int | None = Field(default=0)
    opened_incident_id: UUID | None = None

`HeartbeatPaginatedSchema` ¶

Bases: BasePaginatedSchema

Schema for paginated heartbeat list response.

Attributes:

Name	Type	Description
`data`	`list[HeartbeatListSchema]`	List of HeartbeatListSchema items

Source code in upassist/entities/heartbeat/schemas.py

class HeartbeatPaginatedSchema(BasePaginatedSchema):
    """Schema for paginated heartbeat list response.

    Attributes:
        data: List of HeartbeatListSchema items
    """

    data: list[HeartbeatListSchema]

For more information and to get started with monitoring your applications, visit Upassist Cloud

Heartbeat¶

Features¶

Examples¶

Basic Usage¶

Advanced Configuration¶

Managing Heartbeats¶

API Reference¶

Heartbeat ¶

base_heartbeats_api_url property ¶

__init__(heartbeat_slug=None, api_key=None, api_version=None, api_client_cls=SyncAPIClient) ¶

create(heartbeat) ¶

delete() ¶

detail() ¶

event() ¶

list(q=None, page=None, per_page=None) ¶

pause() ¶

unpause() ¶

HeartbeatCreateSchema ¶

HeartbeatDetailSchema ¶

HeartbeatListSchema ¶

HeartbeatPaginatedSchema ¶

`Heartbeat` ¶

`base_heartbeats_api_url` `property` ¶

`init(heartbeat_slug=None, api_key=None, api_version=None, api_client_cls=SyncAPIClient)` ¶

`create(heartbeat)` ¶

`delete()` ¶

`detail()` ¶

`event()` ¶

`list(q=None, page=None, per_page=None)` ¶

`pause()` ¶

`unpause()` ¶

`HeartbeatCreateSchema` ¶

`HeartbeatDetailSchema` ¶

`HeartbeatListSchema` ¶

`HeartbeatPaginatedSchema` ¶