tessl/pypi-datadog
The Datadog Python library provides tools for interacting with Datadog's monitoring platform through HTTP API client functionality, DogStatsD metrics client, and command-line tools.
—
Pending
http-api-client.mddocs/
HTTP API Client
Complete REST API client providing access to all Datadog HTTP endpoints. The API client handles authentication, request formatting, response parsing, error handling, and retry logic for all Datadog platform functionality.
Capabilities
Events Management
Create, query, and retrieve events in your Datadog event stream for tracking deployments, alerts, and significant occurrences.
class Event:
@staticmethod
def create(title, text, **kwargs):
"""
Post an event to Datadog.
Parameters:
- title (str): Event title
- text (str): Event description
- date_happened (int, optional): Unix timestamp when event occurred
- priority (str, optional): Event priority ('normal' or 'low')
- tags (list, optional): List of tags to apply
- alert_type (str, optional): 'error', 'warning', 'info', or 'success'
- aggregation_key (str, optional): Key for event aggregation
- source_type_name (str, optional): Source type name
- host (str, optional): Host name
Returns:
dict: Response with event details and status
"""
@staticmethod
def query(**kwargs):
"""
Query the event stream.
Parameters:
- start (int): Unix timestamp for start time
- end (int): Unix timestamp for end time
- priority (str, optional): Filter by priority
- sources (str, optional): Comma-separated source list
- tags (str, optional): Comma-separated tag list
- unaggregated (bool, optional): Get individual events
Returns:
dict: Events matching query criteria
"""
@staticmethod
def get(event_id):
"""
Get details for a specific event.
Parameters:
- event_id (int): Event identifier
Returns:
dict: Event details
"""Metrics Management
Query historical metrics data and submit custom metrics to Datadog for monitoring and alerting.
class Metric:
@staticmethod
def query(**kwargs):
"""
Query for metrics data from Datadog.
Parameters:
- query (str): Metrics query string
- start (int): Start time as Unix timestamp
- end (int): End time as Unix timestamp
Returns:
dict: Time series data for the metrics query
"""
@staticmethod
def send(**kwargs):
"""
Submit metrics to Datadog.
Parameters:
- metric (str): Metric name
- points (list): List of [timestamp, value] pairs
- host (str, optional): Host name
- tags (list, optional): List of tags
- type (str, optional): Metric type ('gauge', 'count', 'rate')
- interval (int, optional): Interval in seconds
Returns:
dict: Submission status
"""
@staticmethod
def list(from_epoch):
"""
List all active metrics since a given time.
Parameters:
- from_epoch (int): Start time as Unix timestamp (seconds)
Returns:
dict: List of active metric names
"""Monitors Management
Create, configure, and manage monitors for automated alerting on metrics, logs, traces, and synthetic tests.
class Monitor:
@staticmethod
def create(type, query, **kwargs):
"""
Create a new monitor.
Parameters:
- type (str): Monitor type ('metric alert', 'service check', etc.)
- query (str): Monitor query defining alert conditions
- name (str, optional): Monitor name
- message (str, optional): Alert message with notification details
- tags (list, optional): List of tags
- options (dict, optional): Monitor configuration options
- priority (int, optional): Priority level (1-5)
- restricted_roles (list, optional): Role UUIDs with edit access
Returns:
dict: Created monitor details with ID
"""
@staticmethod
def get_all(**kwargs):
"""
Get all monitors.
Parameters:
- group_states (list, optional): Filter by group states
- name (str, optional): Filter by monitor name
- tags (str, optional): Comma-separated tag filters
- monitor_tags (str, optional): Comma-separated monitor tag filters
- with_downtimes (bool, optional): Include downtime info
- id_offset (int, optional): Monitor ID offset for pagination
- page (int, optional): Page number for pagination
- page_size (int, optional): Number of monitors per page
Returns:
list: List of monitor objects
"""
@staticmethod
def get(monitor_id, **kwargs):
"""
Get monitor details by ID.
Parameters:
- monitor_id (int): Monitor identifier
- group_states (list, optional): Filter by group states
- with_downtimes (bool, optional): Include associated downtimes
Returns:
dict: Monitor configuration and status
"""
@staticmethod
def update(monitor_id, **kwargs):
"""
Update existing monitor.
Parameters:
- monitor_id (int): Monitor identifier
- type (str, optional): Monitor type
- query (str, optional): Updated query
- name (str, optional): Updated name
- message (str, optional): Updated message
- tags (list, optional): Updated tags
- options (dict, optional): Updated options
Returns:
dict: Updated monitor details
"""
@staticmethod
def delete(monitor_id, **kwargs):
"""
Delete a monitor.
Parameters:
- monitor_id (int): Monitor identifier
- force (bool, optional): Force deletion even if referenced
Returns:
dict: Deletion status
"""
@staticmethod
def mute(monitor_id, **kwargs):
"""
Mute monitor notifications.
Parameters:
- monitor_id (int): Monitor identifier
- scope (str, optional): Scope for muting (e.g., 'host:web01')
- end (int, optional): Unix timestamp when muting ends
Returns:
dict: Muting status
"""
@staticmethod
def unmute(monitor_id, **kwargs):
"""
Unmute monitor notifications.
Parameters:
- monitor_id (int): Monitor identifier
- scope (str, optional): Scope for unmuting
Returns:
dict: Unmuting status
"""
@staticmethod
def mute_all():
"""
Globally mute all monitors.
Returns:
dict: Global muting status
"""
@staticmethod
def unmute_all():
"""
Cancel global monitor mute setting (does not remove individual monitor mutes).
Returns:
dict: Global unmuting status
"""
@staticmethod
def search(**kwargs):
"""
Search monitors.
Parameters:
- query (str, optional): Search query for monitors
- page (int, optional): Page number for pagination
- per_page (int, optional): Number of results per page
Returns:
dict: Monitors matching search criteria
"""
@staticmethod
def search_groups(**kwargs):
"""
Search monitor groups.
Parameters:
- query (str, optional): Search query for groups
- page (int, optional): Page number for pagination
- per_page (int, optional): Number of results per page
Returns:
dict: Monitor groups matching search criteria
"""
@staticmethod
def can_delete(**kwargs):
"""
Check if monitors can be deleted.
Parameters:
- monitor_ids (list): List of monitor IDs to check
Returns:
dict: Deletion eligibility status for each monitor
"""
@staticmethod
def validate(**kwargs):
"""
Validate monitor configuration without creating.
Parameters:
- type (str): Monitor type
- query (str): Monitor query
- name (str, optional): Monitor name
- message (str, optional): Alert message
- options (dict, optional): Monitor options
Returns:
dict: Validation results and any errors
"""Dashboard Management
Create and manage dashboards for data visualization with widgets, layouts, and sharing configurations.
class Dashboard:
@staticmethod
def create(**kwargs):
"""
Create a new dashboard.
Parameters:
- title (str): Dashboard title
- widgets (list): List of widget definitions
- layout_type (str): Layout type ('ordered' or 'free')
- description (str, optional): Dashboard description
- is_read_only (bool, optional): Read-only status
- notify_list (list, optional): List of notification handles
- template_variables (list, optional): Template variable definitions
- reflow_type (str, optional): Reflow behavior for 'ordered' layout
Returns:
dict: Created dashboard with ID and URL
"""
@staticmethod
def get_all(**kwargs):
"""
Get all dashboards.
Parameters:
- filter_shared (bool, optional): Filter by shared status
- filter_deleted (bool, optional): Include deleted dashboards
Returns:
dict: List of dashboard summaries
"""
@staticmethod
def get(dashboard_id):
"""
Get dashboard by ID.
Parameters:
- dashboard_id (str): Dashboard identifier
Returns:
dict: Complete dashboard configuration
"""
@staticmethod
def update(dashboard_id, **kwargs):
"""
Update existing dashboard.
Parameters:
- dashboard_id (str): Dashboard identifier
- title (str, optional): Updated title
- widgets (list, optional): Updated widget list
- layout_type (str, optional): Updated layout type
- description (str, optional): Updated description
Returns:
dict: Updated dashboard details
"""
@staticmethod
def delete(dashboard_id):
"""
Delete a dashboard.
Parameters:
- dashboard_id (str): Dashboard identifier
Returns:
dict: Deletion confirmation
"""Infrastructure Management
Manage hosts, tags, and infrastructure components for monitoring and organization.
class Host:
@staticmethod
def mute(hostname, **kwargs):
"""
Mute notifications for a host.
Parameters:
- hostname (str): Host name to mute
- end (int, optional): Unix timestamp when muting ends
- message (str, optional): Muting reason message
- override (bool, optional): Override existing muting
Returns:
dict: Muting status and details
"""
@staticmethod
def unmute(hostname):
"""
Unmute notifications for a host.
Parameters:
- hostname (str): Host name to unmute
Returns:
dict: Unmuting confirmation
"""
class Hosts:
@staticmethod
def search(**kwargs):
"""
Search for hosts.
Parameters:
- filter (str, optional): Search filter
- sort_field (str, optional): Field to sort by
- sort_dir (str, optional): Sort direction ('asc' or 'desc')
- start (int, optional): Start index for pagination
- count (int, optional): Number of results to return
- include_muted_hosts_data (bool, optional): Include muted hosts
- include_hosts_metadata (bool, optional): Include metadata
Returns:
dict: Host search results with metadata
"""
@staticmethod
def get_all():
"""
Get all hosts.
Returns:
list: List of all host names
"""
@staticmethod
def totals():
"""
Get host count totals.
Returns:
dict: Total host counts by status
"""
class Tag:
@staticmethod
def create(host, tags, **kwargs):
"""
Add tags to a host.
Parameters:
- host (str): Host name
- tags (list): List of tags to add
- source (str, optional): Tag source identifier
Returns:
dict: Updated host tags
"""
@staticmethod
def update(host, tags, **kwargs):
"""
Update host tags (replaces existing).
Parameters:
- host (str): Host name
- tags (list): Complete list of tags
- source (str, optional): Tag source identifier
Returns:
dict: Updated host tags
"""
@staticmethod
def get(host, **kwargs):
"""
Get tags for a host.
Parameters:
- host (str): Host name
- source (str, optional): Filter by tag source
- by_source (bool, optional): Group tags by source
Returns:
dict: Host tags organized by source
"""
@staticmethod
def get_all(**kwargs):
"""
Get all tags.
Parameters:
- source (str, optional): Filter by tag source
Returns:
dict: All tags in the organization
"""
@staticmethod
def delete(host, **kwargs):
"""
Remove all tags from a host.
Parameters:
- host (str): Host name
- source (str, optional): Remove tags from specific source only
Returns:
dict: Deletion confirmation
"""Downtime Management
Schedule and manage downtime periods to suppress monitor notifications during maintenance windows.
class Downtime:
@staticmethod
def create(**kwargs):
"""
Schedule downtime.
Parameters:
- scope (list): Downtime scope (host tags, monitor tags, etc.)
- start (int, optional): Unix timestamp for start time
- end (int, optional): Unix timestamp for end time
- message (str, optional): Downtime message
- timezone (str, optional): Timezone identifier
- recurrence (dict, optional): Recurrence configuration
- monitor_id (int, optional): Specific monitor to mute
- monitor_tags (list, optional): Monitor tags to match
Returns:
dict: Created downtime with ID and schedule
"""
@staticmethod
def get_all(**kwargs):
"""
Get all downtimes.
Parameters:
- current_only (bool, optional): Only active downtimes
- with_creator (bool, optional): Include creator information
Returns:
list: List of scheduled downtimes
"""
@staticmethod
def get(downtime_id):
"""
Get downtime by ID.
Parameters:
- downtime_id (int): Downtime identifier
Returns:
dict: Downtime details and status
"""
@staticmethod
def update(downtime_id, **kwargs):
"""
Update existing downtime.
Parameters:
- downtime_id (int): Downtime identifier
- scope (list, optional): Updated scope
- start (int, optional): Updated start time
- end (int, optional): Updated end time
- message (str, optional): Updated message
Returns:
dict: Updated downtime details
"""
@staticmethod
def delete(downtime_id):
"""
Cancel scheduled downtime.
Parameters:
- downtime_id (int): Downtime identifier
Returns:
dict: Cancellation confirmation
"""Service Level Objectives
Define, track, and manage SLOs for measuring service reliability and performance targets.
class ServiceLevelObjective:
@staticmethod
def create(**kwargs):
"""
Create a new SLO.
Parameters:
- name (str): SLO name
- type (str): SLO type ('metric' or 'monitor')
- query (dict): Query definition for the SLO
- thresholds (list): List of threshold configurations
- timeframe (str): Time window ('7d', '30d', '90d')
- target_threshold (float): Target percentage (0-100)
- warning_threshold (float, optional): Warning percentage
- description (str, optional): SLO description
- tags (list, optional): List of tags
Returns:
dict: Created SLO with ID and configuration
"""
@staticmethod
def get_all(**kwargs):
"""
Get all SLOs.
Parameters:
- ids (str, optional): Comma-separated SLO IDs
- query (str, optional): Search query
- tags_query (str, optional): Tags filter query
- metrics_query (str, optional): Metrics filter query
- limit (int, optional): Maximum number of results
- offset (int, optional): Offset for pagination
Returns:
dict: List of SLO summaries
"""
@staticmethod
def get(slo_id, **kwargs):
"""
Get SLO by ID.
Parameters:
- slo_id (str): SLO identifier
- with_configured_alert_ids (bool, optional): Include alert IDs
Returns:
dict: Complete SLO configuration and status
"""
@staticmethod
def update(slo_id, **kwargs):
"""
Update existing SLO.
Parameters:
- slo_id (str): SLO identifier
- name (str, optional): Updated name
- query (dict, optional): Updated query
- thresholds (list, optional): Updated thresholds
- target_threshold (float, optional): Updated target
- description (str, optional): Updated description
Returns:
dict: Updated SLO details
"""
@staticmethod
def delete(slo_id, **kwargs):
"""
Delete an SLO.
Parameters:
- slo_id (str): SLO identifier
- force (bool, optional): Force deletion if referenced
Returns:
dict: Deletion confirmation
"""
@staticmethod
def history(slo_id, **kwargs):
"""
Get SLO history data.
Parameters:
- slo_id (str): SLO identifier
- from_ts (int): Start time as Unix timestamp
- to_ts (int): End time as Unix timestamp
- target (float, optional): Target threshold for calculations
- apply_correction (bool, optional): Apply data corrections
Returns:
dict: Historical SLO performance data
"""User and Role Management
Manage users, custom roles, and permissions for organization access control and security.
class User:
@staticmethod
def create(**kwargs):
"""
Create a new user.
Parameters:
- email (str): User email address
- handle (str): User handle/username
- name (str): User full name
- access_role (str, optional): User access role
- disabled (bool, optional): Account disabled status
Returns:
dict: Created user details
"""
@staticmethod
def get_all():
"""
Get all users.
Returns:
list: List of all users in organization
"""
@staticmethod
def get(handle):
"""
Get user by handle.
Parameters:
- handle (str): User handle/username
Returns:
dict: User details and settings
"""
@staticmethod
def update(handle, **kwargs):
"""
Update user information.
Parameters:
- handle (str): User handle/username
- email (str, optional): Updated email
- name (str, optional): Updated name
- access_role (str, optional): Updated access role
- disabled (bool, optional): Updated disabled status
Returns:
dict: Updated user details
"""
@staticmethod
def delete(handle):
"""
Delete a user.
Parameters:
- handle (str): User handle/username
Returns:
dict: Deletion confirmation
"""
class Roles:
@staticmethod
def create(**kwargs):
"""
Create a custom role.
Parameters:
- name (str): Role name
- permissions (list, optional): List of permission IDs
Returns:
dict: Created role with ID
"""
@staticmethod
def get_all():
"""
Get all roles.
Returns:
list: List of all roles including built-in
"""
@staticmethod
def get(role_id):
"""
Get role by ID.
Parameters:
- role_id (str): Role identifier
Returns:
dict: Role details and permissions
"""
@staticmethod
def update(role_id, **kwargs):
"""
Update role configuration.
Parameters:
- role_id (str): Role identifier
- name (str, optional): Updated role name
- permissions (list, optional): Updated permission list
Returns:
dict: Updated role details
"""
@staticmethod
def delete(role_id):
"""
Delete a custom role.
Parameters:
- role_id (str): Role identifier
Returns:
dict: Deletion confirmation
"""
class Permissions:
@staticmethod
def get_all():
"""
Get all available permissions.
Returns:
list: List of all permission objects with IDs and descriptions
"""Cloud Integrations
Configure and manage cloud provider integrations for automatic resource discovery and monitoring.
class AwsIntegration:
@staticmethod
def create(**kwargs):
"""
Create AWS integration.
Parameters:
- account_id (str): AWS account ID
- role_name (str): IAM role name for Datadog
- filter_tags (list, optional): Resource filtering tags
- host_tags (list, optional): Tags to apply to hosts
- account_specific_namespace_rules (dict, optional): Namespace rules
- excluded_regions (list, optional): Regions to exclude
- cspm_resource_collection_enabled (bool, optional): Enable CSPM
- metrics_collection_enabled (bool, optional): Enable metrics
Returns:
dict: Created integration configuration
"""
@staticmethod
def get_all():
"""
Get all AWS integrations.
Returns:
list: List of AWS integration configurations
"""
@staticmethod
def update(account_id, role_name, **kwargs):
"""
Update AWS integration.
Parameters:
- account_id (str): AWS account ID
- role_name (str): IAM role name
- filter_tags (list, optional): Updated filtering tags
- host_tags (list, optional): Updated host tags
Returns:
dict: Updated integration configuration
"""
@staticmethod
def delete(account_id, role_name):
"""
Delete AWS integration.
Parameters:
- account_id (str): AWS account ID
- role_name (str): IAM role name
Returns:
dict: Deletion confirmation
"""
class AzureIntegration:
@staticmethod
def create(**kwargs):
"""
Create Azure integration.
Parameters:
- tenant_name (str): Azure tenant name
- client_id (str): Azure client ID
- client_secret (str): Azure client secret
- host_filters (str, optional): Host filtering expression
- app_service_plan_filters (str, optional): App Service filters
- container_app_filters (str, optional): Container app filters
Returns:
dict: Created Azure integration
"""
@staticmethod
def get_all():
"""
Get all Azure integrations.
Returns:
list: List of Azure integration configurations
"""
class GcpIntegration:
@staticmethod
def create(**kwargs):
"""
Create GCP integration.
Parameters:
- project_id (str): GCP project ID
- private_key_id (str): Service account private key ID
- private_key (str): Service account private key
- client_email (str): Service account email
- client_id (str): Service account client ID
- host_filters (str, optional): Host filtering expression
- resource_collection_enabled (bool, optional): Enable resource collection
Returns:
dict: Created GCP integration
"""
@staticmethod
def get_all():
"""
Get all GCP integrations.
Returns:
list: List of GCP integration configurations
"""Synthetic Monitoring
Create and manage synthetic tests for proactive monitoring of web applications, APIs, and user journeys.
class Synthetics:
@staticmethod
def create_test(**kwargs):
"""
Create a synthetic test.
Parameters:
- type (str): Test type ('api', 'browser', 'mobile')
- config (dict): Test configuration with request details
- options (dict): Test options and settings
- locations (list): List of test locations
- message (str): Alert message for test failures
- name (str): Test name
- tags (list, optional): List of tags
- status (str, optional): Test status ('live' or 'paused')
Returns:
dict: Created test with ID and configuration
"""
@staticmethod
def get_all_tests(**kwargs):
"""
Get all synthetic tests.
Returns:
list: List of all synthetic tests
"""
@staticmethod
def get_test(public_id):
"""
Get synthetic test by ID.
Parameters:
- public_id (str): Test public identifier
Returns:
dict: Complete test configuration
"""
@staticmethod
def update_test(public_id, **kwargs):
"""
Update synthetic test.
Parameters:
- public_id (str): Test public identifier
- name (str, optional): Updated test name
- config (dict, optional): Updated configuration
- options (dict, optional): Updated options
- locations (list, optional): Updated locations
Returns:
dict: Updated test configuration
"""
@staticmethod
def delete_test(public_id):
"""
Delete synthetic test.
Parameters:
- public_id (str): Test public identifier
Returns:
dict: Deletion confirmation
"""Usage Examples
Creating and Managing Monitors
from datadog import initialize, api
# Initialize the client
initialize(api_key="your-api-key", app_key="your-app-key")
# Create a metric monitor
monitor = api.Monitor.create(
type="metric alert",
query="avg(last_5m):avg:system.cpu.user{*} > 80",
name="High CPU usage",
message="CPU usage is above 80% @webhook-http://example.com/hook",
tags=["environment:production", "team:infrastructure"],
options={
"thresholds": {"critical": 80, "warning": 70},
"new_host_delay": 300,
"evaluation_delay": 900,
"no_data_timeframe": 10,
"renotify_interval": 60
}
)
# Get monitor details
monitor_details = api.Monitor.get(monitor['id'])
# Update monitor
api.Monitor.update(
monitor['id'],
message="Updated alert message with new webhook @webhook-slack-alerts"
)
# Mute monitor temporarily
api.Monitor.mute(monitor['id'], end=int(time.time()) + 3600) # 1 hour
# Clean up - delete monitor
api.Monitor.delete(monitor['id'])Dashboard Creation with Widgets
# Create a dashboard with multiple widgets
dashboard = api.Dashboard.create(
title="Application Performance Dashboard",
layout_type="ordered",
widgets=[
{
"definition": {
"type": "timeseries",
"requests": [
{
"q": "avg:system.cpu.user{environment:production}",
"display_type": "line",
"style": {"palette": "dog_classic", "line_type": "solid", "line_width": "normal"}
}
],
"title": "CPU Usage Over Time",
"yaxis": {"min": "0", "max": "100"}
}
},
{
"definition": {
"type": "query_value",
"requests": [
{
"q": "avg:system.load.1{environment:production}",
"aggregator": "avg"
}
],
"title": "Current Load Average",
"precision": 2
}
}
],
description="Real-time monitoring for production application performance",
template_variables=[
{
"name": "environment",
"default": "production",
"prefix": "environment"
}
]
)
print(f"Dashboard created: {dashboard['url']}")Install with Tessl CLI
npx tessl i tessl/pypi-datadog