How to Use Python with Datadog: SDKs, Metrics, Logs, and Product-Specific Data
READER BEWARE: THE FOLLOWING WRITTEN ENTIRELY BY AI WITHOUT HUMAN EDITING.
Introduction
Datadog is a powerful monitoring and analytics platform that helps organizations observe their applications and infrastructure. When working with Datadog, Python developers have multiple options for integration: using official SDKs or making direct API calls. This guide explores both approaches, comparing different SDKs, demonstrating authentication methods, and showing how to retrieve metrics, logs, and product-specific data.
Datadog Python SDKs: An Overview
Datadog provides several Python SDKs for different purposes. Understanding which SDK to use is crucial for effective integration.
1. Official Datadog API Client (datadog-api-client)
The datadog-api-client is the official, auto-generated Python client library for the Datadog API. It provides comprehensive access to all Datadog API endpoints.
Key Features:
- Full API coverage (v1 and v2 endpoints)
- Auto-generated from OpenAPI specifications
- Type hints for better IDE support
- Synchronous operations
- Official support from Datadog
Installation:
pip install datadog-api-client
Official Documentation: https://docs.datadoghq.com/api/latest/
2. DogStatsD Client (datadogpy)
The datadogpy library includes the DogStatsD client for sending custom metrics, events, and service checks to Datadog.
Key Features:
- Lightweight metrics submission
- UDP-based for minimal performance impact
- Buffered sending
- Custom metrics and events
Installation:
pip install datadog
Official Documentation: https://datadogpy.readthedocs.io/
3. Datadog Lambda Layer (for AWS Lambda)
For serverless applications, Datadog provides Lambda layers that include Python libraries for tracing and monitoring.
Official Documentation: https://docs.datadoghq.com/serverless/
SDK Comparison Table
| SDK | Purpose | Best For | Installation |
|---|---|---|---|
datadog-api-client | Full API access | Querying metrics, logs, dashboards; programmatic management | pip install datadog-api-client |
datadogpy (DogStatsD) | Metrics submission | Sending custom metrics from applications | pip install datadog |
APM Tracer (ddtrace) | Application tracing | Distributed tracing and APM | pip install ddtrace |
| Lambda Layer | Serverless monitoring | AWS Lambda functions | Via AWS Lambda Layer |
Authentication Methods
Both SDKs and direct API calls require authentication using API keys and Application keys.
API Keys vs Application Keys
- API Key: Required for all API requests. Used to identify your organization. Can be found in Datadog UI under Organization Settings → API Keys.
- Application Key: Required for querying and managing resources (metrics, logs, dashboards). More granular permissions. Found under Organization Settings → Application Keys.
Setting Up Authentication
Environment Variables (Recommended):
export DD_API_KEY="your_api_key_here"
export DD_APP_KEY="your_app_key_here"
export DD_SITE="datadoghq.com" # or datadoghq.eu, us3.datadoghq.com, etc.
In Python Configuration:
import os
# Load from environment variables
api_key = os.getenv('DD_API_KEY')
app_key = os.getenv('DD_APP_KEY')
dd_site = os.getenv('DD_SITE', 'datadoghq.com')
Using the Datadog API Client SDK
Let’s explore how to use the official datadog-api-client SDK for various operations.
Installation and Setup
from datadog_api_client import ApiClient, Configuration
from datadog_api_client.v1.api.metrics_api import MetricsApi
from datadog_api_client.v2.api.logs_api import LogsApi
from datadog_api_client.v1.api.dashboards_api import DashboardsApi
import os
# Configure the API client
configuration = Configuration()
configuration.api_key['apiKeyAuth'] = os.getenv('DD_API_KEY')
configuration.api_key['appKeyAuth'] = os.getenv('DD_APP_KEY')
configuration.server_variables['site'] = os.getenv('DD_SITE', 'datadoghq.com')
Retrieving Metrics with SDK
from datadog_api_client import ApiClient, Configuration
from datadog_api_client.v1.api.metrics_api import MetricsApi
from datadog_api_client.v1.model.metrics_query_response import MetricsQueryResponse
import os
import time
def query_metrics_sdk():
"""Query metrics using the Datadog API client SDK."""
# Configure authentication
configuration = Configuration()
configuration.api_key['apiKeyAuth'] = os.getenv('DD_API_KEY')
configuration.api_key['appKeyAuth'] = os.getenv('DD_APP_KEY')
configuration.server_variables['site'] = os.getenv('DD_SITE', 'datadoghq.com')
# Create API client
with ApiClient(configuration) as api_client:
api_instance = MetricsApi(api_client)
# Define time range (last 1 hour)
end_time = int(time.time())
start_time = end_time - 3600
# Query metrics
query = "avg:system.cpu.user{*}"
try:
response = api_instance.query_metrics(
_from=start_time,
to=end_time,
query=query
)
print(f"Query: {query}")
print(f"Status: {response.status}")
if response.series:
for series in response.series:
print(f"\nMetric: {series.metric}")
print(f"Scope: {series.scope}")
print(f"Display Name: {series.display_name}")
print(f"Points: {len(series.pointlist)} data points")
# Show first 5 points
for point in series.pointlist[:5]:
timestamp, value = point
print(f" {timestamp}: {value}")
else:
print("No data returned")
return response
except Exception as e:
print(f"Error querying metrics: {e}")
raise
# Example usage
if __name__ == "__main__":
query_metrics_sdk()
Retrieving Logs with SDK
from datadog_api_client import ApiClient, Configuration
from datadog_api_client.v2.api.logs_api import LogsApi
from datadog_api_client.v2.model.logs_list_request import LogsListRequest
from datadog_api_client.v2.model.logs_query_filter import LogsQueryFilter
from datadog_api_client.v2.model.logs_sort import LogsSort
from datetime import datetime, timedelta
import os
def query_logs_sdk():
"""Query logs using the Datadog API client SDK."""
# Configure authentication
configuration = Configuration()
configuration.api_key['apiKeyAuth'] = os.getenv('DD_API_KEY')
configuration.api_key['appKeyAuth'] = os.getenv('DD_APP_KEY')
configuration.server_variables['site'] = os.getenv('DD_SITE', 'datadoghq.com')
# Create API client
with ApiClient(configuration) as api_client:
api_instance = LogsApi(api_client)
# Define time range (last 15 minutes)
end_time = datetime.utcnow()
start_time = end_time - timedelta(minutes=15)
# Create logs query request
body = LogsListRequest(
filter=LogsQueryFilter(
query="service:my-service status:error",
_from=start_time.isoformat() + "Z",
to=end_time.isoformat() + "Z",
),
sort=LogsSort.TIMESTAMP_ASCENDING,
page=dict(limit=100)
)
try:
response = api_instance.list_logs(body=body)
print(f"Retrieved {len(response.data)} log entries")
for log in response.data:
attributes = log.attributes
print(f"\nTimestamp: {attributes.timestamp}")
print(f"Status: {attributes.status}")
print(f"Service: {attributes.get('service', 'N/A')}")
print(f"Message: {attributes.get('message', 'N/A')[:100]}")
# Show tags if available
if attributes.tags:
print(f"Tags: {', '.join(attributes.tags[:5])}")
return response
except Exception as e:
print(f"Error querying logs: {e}")
raise
# Example usage
if __name__ == "__main__":
query_logs_sdk()
Retrieving Product-Specific Data with SDK
Datadog’s API provides access to various product-specific data like dashboards, monitors, SLOs, and more.
from datadog_api_client import ApiClient, Configuration
from datadog_api_client.v1.api.dashboards_api import DashboardsApi
from datadog_api_client.v1.api.monitors_api import MonitorsApi
from datadog_api_client.v1.api.service_level_objectives_api import ServiceLevelObjectivesApi
import os
def get_dashboards_sdk():
"""Retrieve dashboards using the Datadog API client SDK."""
# Configure authentication
configuration = Configuration()
configuration.api_key['apiKeyAuth'] = os.getenv('DD_API_KEY')
configuration.api_key['appKeyAuth'] = os.getenv('DD_APP_KEY')
configuration.server_variables['site'] = os.getenv('DD_SITE', 'datadoghq.com')
with ApiClient(configuration) as api_client:
api_instance = DashboardsApi(api_client)
try:
# List all dashboards
response = api_instance.list_dashboards()
print(f"Found {len(response.dashboards)} dashboards:")
for dashboard in response.dashboards[:10]: # Show first 10
print(f"\n ID: {dashboard.id}")
print(f" Title: {dashboard.title}")
print(f" Description: {dashboard.get('description', 'N/A')}")
print(f" Created: {dashboard.created}")
print(f" Modified: {dashboard.modified}")
return response
except Exception as e:
print(f"Error retrieving dashboards: {e}")
raise
def get_monitors_sdk():
"""Retrieve monitors using the Datadog API client SDK."""
# Configure authentication
configuration = Configuration()
configuration.api_key['apiKeyAuth'] = os.getenv('DD_API_KEY')
configuration.api_key['appKeyAuth'] = os.getenv('DD_APP_KEY')
configuration.server_variables['site'] = os.getenv('DD_SITE', 'datadoghq.com')
with ApiClient(configuration) as api_client:
api_instance = MonitorsApi(api_client)
try:
# List monitors with filtering
response = api_instance.list_monitors(
group_states="alert,warn,no data",
tags="service:my-service"
)
print(f"Found {len(response)} monitors:")
for monitor in response[:10]: # Show first 10
print(f"\n ID: {monitor.id}")
print(f" Name: {monitor.name}")
print(f" Type: {monitor.type}")
print(f" Status: {monitor.overall_state}")
print(f" Tags: {monitor.tags}")
return response
except Exception as e:
print(f"Error retrieving monitors: {e}")
raise
def get_slos_sdk():
"""Retrieve Service Level Objectives using the Datadog API client SDK."""
# Configure authentication
configuration = Configuration()
configuration.api_key['apiKeyAuth'] = os.getenv('DD_API_KEY')
configuration.api_key['appKeyAuth'] = os.getenv('DD_APP_KEY')
configuration.server_variables['site'] = os.getenv('DD_SITE', 'datadoghq.com')
with ApiClient(configuration) as api_client:
api_instance = ServiceLevelObjectivesApi(api_client)
try:
# List all SLOs
response = api_instance.list_slos()
print(f"Found {len(response.data)} SLOs:")
for slo in response.data[:10]: # Show first 10
print(f"\n ID: {slo.id}")
print(f" Name: {slo.name}")
print(f" Type: {slo.type}")
print(f" Target: {slo.target_threshold}%")
print(f" Tags: {slo.tags}")
return response
except Exception as e:
print(f"Error retrieving SLOs: {e}")
raise
# Example usage
if __name__ == "__main__":
print("=== Dashboards ===")
get_dashboards_sdk()
print("\n=== Monitors ===")
get_monitors_sdk()
print("\n=== SLOs ===")
get_slos_sdk()
Using the Datadog API Directly
While SDKs provide convenience, sometimes you may want to use the REST API directly using Python’s requests library. This approach gives you complete control and doesn’t require SDK dependencies.
Authentication with Direct API Calls
import requests
import os
class DatadogAPIClient:
"""Simple wrapper for Datadog API calls."""
def __init__(self):
self.api_key = os.getenv('DD_API_KEY')
self.app_key = os.getenv('DD_APP_KEY')
self.site = os.getenv('DD_SITE', 'datadoghq.com')
self.base_url = f"https://api.{self.site}"
# Validate credentials
if not self.api_key or not self.app_key:
raise ValueError("DD_API_KEY and DD_APP_KEY must be set")
def _get_headers(self):
"""Get authentication headers for API requests."""
return {
'DD-API-KEY': self.api_key,
'DD-APPLICATION-KEY': self.app_key,
'Content-Type': 'application/json'
}
def get(self, endpoint, params=None):
"""Make a GET request to the Datadog API."""
url = f"{self.base_url}{endpoint}"
response = requests.get(url, headers=self._get_headers(), params=params)
response.raise_for_status()
return response.json()
def post(self, endpoint, data=None):
"""Make a POST request to the Datadog API."""
url = f"{self.base_url}{endpoint}"
response = requests.post(url, headers=self._get_headers(), json=data)
response.raise_for_status()
return response.json()
Retrieving Metrics with Direct API
import requests
import time
import os
def query_metrics_api():
"""Query metrics using direct Datadog API calls."""
# Authentication
api_key = os.getenv('DD_API_KEY')
app_key = os.getenv('DD_APP_KEY')
dd_site = os.getenv('DD_SITE', 'datadoghq.com')
# API endpoint
url = f"https://api.{dd_site}/api/v1/query"
# Headers
headers = {
'DD-API-KEY': api_key,
'DD-APPLICATION-KEY': app_key,
'Content-Type': 'application/json'
}
# Define time range (last 1 hour)
end_time = int(time.time())
start_time = end_time - 3600
# Query parameters
params = {
'query': 'avg:system.cpu.user{*}',
'from': start_time,
'to': end_time
}
try:
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
data = response.json()
print(f"Query: {params['query']}")
print(f"Status: {data.get('status')}")
if data.get('series'):
for series in data['series']:
print(f"\nMetric: {series['metric']}")
print(f"Scope: {series['scope']}")
print(f"Display Name: {series['display_name']}")
print(f"Points: {len(series['pointlist'])} data points")
# Show first 5 points
for point in series['pointlist'][:5]:
timestamp, value = point
print(f" {timestamp}: {value}")
else:
print("No data returned")
return data
except requests.exceptions.RequestException as e:
print(f"Error querying metrics: {e}")
if hasattr(e.response, 'text'):
print(f"Response: {e.response.text}")
raise
# Example usage
if __name__ == "__main__":
query_metrics_api()
Retrieving Logs with Direct API
import requests
from datetime import datetime, timedelta
import os
def query_logs_api():
"""Query logs using direct Datadog API calls."""
# Authentication
api_key = os.getenv('DD_API_KEY')
app_key = os.getenv('DD_APP_KEY')
dd_site = os.getenv('DD_SITE', 'datadoghq.com')
# API endpoint (v2)
url = f"https://api.{dd_site}/api/v2/logs/events/search"
# Headers
headers = {
'DD-API-KEY': api_key,
'DD-APPLICATION-KEY': app_key,
'Content-Type': 'application/json'
}
# Define time range (last 15 minutes)
end_time = datetime.utcnow()
start_time = end_time - timedelta(minutes=15)
# Request body
body = {
"filter": {
"query": "service:my-service status:error",
"from": start_time.isoformat() + "Z",
"to": end_time.isoformat() + "Z"
},
"sort": "timestamp",
"page": {
"limit": 100
}
}
try:
response = requests.post(url, headers=headers, json=body)
response.raise_for_status()
data = response.json()
logs = data.get('data', [])
print(f"Retrieved {len(logs)} log entries")
for log in logs:
attributes = log.get('attributes', {})
print(f"\nTimestamp: {attributes.get('timestamp')}")
print(f"Status: {attributes.get('status')}")
print(f"Service: {attributes.get('service', 'N/A')}")
message = attributes.get('message', 'N/A')
print(f"Message: {message[:100]}")
# Show tags if available
tags = attributes.get('tags', [])
if tags:
print(f"Tags: {', '.join(tags[:5])}")
return data
except requests.exceptions.RequestException as e:
print(f"Error querying logs: {e}")
if hasattr(e.response, 'text'):
print(f"Response: {e.response.text}")
raise
# Example usage
if __name__ == "__main__":
query_logs_api()
Retrieving Product-Specific Data with Direct API
import requests
import os
def get_dashboards_api():
"""Retrieve dashboards using direct Datadog API calls."""
# Authentication
api_key = os.getenv('DD_API_KEY')
app_key = os.getenv('DD_APP_KEY')
dd_site = os.getenv('DD_SITE', 'datadoghq.com')
# API endpoint
url = f"https://api.{dd_site}/api/v1/dashboard"
# Headers
headers = {
'DD-API-KEY': api_key,
'DD-APPLICATION-KEY': app_key,
'Content-Type': 'application/json'
}
try:
response = requests.get(url, headers=headers)
response.raise_for_status()
data = response.json()
dashboards = data.get('dashboards', [])
print(f"Found {len(dashboards)} dashboards:")
for dashboard in dashboards[:10]: # Show first 10
print(f"\n ID: {dashboard.get('id')}")
print(f" Title: {dashboard.get('title')}")
print(f" Description: {dashboard.get('description', 'N/A')}")
print(f" Created: {dashboard.get('created')}")
print(f" Modified: {dashboard.get('modified')}")
return data
except requests.exceptions.RequestException as e:
print(f"Error retrieving dashboards: {e}")
if hasattr(e.response, 'text'):
print(f"Response: {e.response.text}")
raise
def get_monitors_api():
"""Retrieve monitors using direct Datadog API calls."""
# Authentication
api_key = os.getenv('DD_API_KEY')
app_key = os.getenv('DD_APP_KEY')
dd_site = os.getenv('DD_SITE', 'datadoghq.com')
# API endpoint
url = f"https://api.{dd_site}/api/v1/monitor"
# Headers
headers = {
'DD-API-KEY': api_key,
'DD-APPLICATION-KEY': app_key,
'Content-Type': 'application/json'
}
# Query parameters
params = {
'group_states': 'alert,warn,no data',
'tags': 'service:my-service'
}
try:
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
monitors = response.json()
print(f"Found {len(monitors)} monitors:")
for monitor in monitors[:10]: # Show first 10
print(f"\n ID: {monitor.get('id')}")
print(f" Name: {monitor.get('name')}")
print(f" Type: {monitor.get('type')}")
print(f" Status: {monitor.get('overall_state')}")
print(f" Tags: {monitor.get('tags')}")
return monitors
except requests.exceptions.RequestException as e:
print(f"Error retrieving monitors: {e}")
if hasattr(e.response, 'text'):
print(f"Response: {e.response.text}")
raise
def get_slos_api():
"""Retrieve Service Level Objectives using direct Datadog API calls."""
# Authentication
api_key = os.getenv('DD_API_KEY')
app_key = os.getenv('DD_APP_KEY')
dd_site = os.getenv('DD_SITE', 'datadoghq.com')
# API endpoint
url = f"https://api.{dd_site}/api/v1/slo"
# Headers
headers = {
'DD-API-KEY': api_key,
'DD-APPLICATION-KEY': app_key,
'Content-Type': 'application/json'
}
try:
response = requests.get(url, headers=headers)
response.raise_for_status()
data = response.json()
slos = data.get('data', [])
print(f"Found {len(slos)} SLOs:")
for slo in slos[:10]: # Show first 10
print(f"\n ID: {slo.get('id')}")
print(f" Name: {slo.get('name')}")
print(f" Type: {slo.get('type')}")
# Handle thresholds based on SLO type
thresholds = slo.get('thresholds', [])
if thresholds:
print(f" Target: {thresholds[0].get('target')}%")
print(f" Tags: {slo.get('tags')}")
return data
except requests.exceptions.RequestException as e:
print(f"Error retrieving SLOs: {e}")
if hasattr(e.response, 'text'):
print(f"Response: {e.response.text}")
raise
# Example usage
if __name__ == "__main__":
print("=== Dashboards ===")
get_dashboards_api()
print("\n=== Monitors ===")
get_monitors_api()
print("\n=== SLOs ===")
get_slos_api()
SDK vs API: Side-by-Side Comparison
Metrics Query Comparison
SDK Approach:
# Pros: Type safety, auto-completion, cleaner code
# Cons: Additional dependency, learning curve
from datadog_api_client import ApiClient, Configuration
from datadog_api_client.v1.api.metrics_api import MetricsApi
configuration = Configuration()
configuration.api_key['apiKeyAuth'] = os.getenv('DD_API_KEY')
configuration.api_key['appKeyAuth'] = os.getenv('DD_APP_KEY')
with ApiClient(configuration) as api_client:
api = MetricsApi(api_client)
response = api.query_metrics(
_from=start_time,
to=end_time,
query="avg:system.cpu.user{*}"
)
Direct API Approach:
# Pros: No additional dependencies, simpler for one-off scripts
# Cons: Manual error handling, no type checking, more verbose
import requests
response = requests.get(
f"https://api.{dd_site}/api/v1/query",
headers={
'DD-API-KEY': api_key,
'DD-APPLICATION-KEY': app_key
},
params={
'query': 'avg:system.cpu.user{*}',
'from': start_time,
'to': end_time
}
)
data = response.json()
Logs Query Comparison
SDK Approach:
# More structured with model objects
from datadog_api_client.v2.api.logs_api import LogsApi
from datadog_api_client.v2.model.logs_list_request import LogsListRequest
api = LogsApi(api_client)
body = LogsListRequest(
filter=LogsQueryFilter(
query="service:my-service status:error",
_from=start_time.isoformat() + "Z",
to=end_time.isoformat() + "Z"
)
)
response = api.list_logs(body=body)
Direct API Approach:
# More flexible, lighter weight
response = requests.post(
f"https://api.{dd_site}/api/v2/logs/events/search",
headers=headers,
json={
"filter": {
"query": "service:my-service status:error",
"from": start_time.isoformat() + "Z",
"to": end_time.isoformat() + "Z"
}
}
)
data = response.json()
When to Use SDK vs Direct API
Use the SDK When:
- Building Production Applications: SDKs provide better error handling and validation
- Type Safety Matters: You want IDE auto-completion and type checking
- Complex Integrations: Managing multiple API endpoints and resources
- Team Collaboration: Standardized approach for team projects
- Long-term Maintenance: SDK updates handle API changes automatically
Use Direct API When:
- Simple Scripts: One-off data extraction or reporting scripts
- Minimal Dependencies: Don’t want to add external packages
- Custom Requirements: Need fine-grained control over requests
- Debugging: Understanding exact API behavior
- Constrained Environments: Limited package installation options
Comparison Table: SDK vs Direct API
| Aspect | SDK (datadog-api-client) | Direct API (requests) |
|---|---|---|
| Setup Complexity | Medium (install package, configure) | Low (just requests library) |
| Code Verbosity | Low (concise, typed) | Medium (more manual work) |
| Type Safety | Yes (type hints, models) | No (raw dictionaries) |
| Error Handling | Built-in exceptions | Manual with status codes |
| Documentation | Generated docs + API docs | API docs only |
| IDE Support | Excellent (auto-completion) | Basic (no type info) |
| Flexibility | High (covers all endpoints) | Very High (complete control) |
| Dependencies | Additional package | Just requests (often pre-installed) |
| Learning Curve | Medium (SDK-specific patterns) | Low (standard HTTP) |
| Version Updates | SDK updates with API | Must update manually |
| Best For | Production apps, complex integrations | Scripts, debugging, simple tools |
Best Practices
Security Best Practices
- Never Hardcode Credentials: Always use environment variables or secrets management
- Rotate Keys Regularly: Implement key rotation policies
- Use Application Keys with Minimum Permissions: Create keys with only necessary scopes
- Secure Storage: Use secret management tools (AWS Secrets Manager, HashiCorp Vault, etc.)
# Good: Using environment variables
api_key = os.getenv('DD_API_KEY')
# Bad: Hardcoding credentials
api_key = "abc123..." # Never do this!
Error Handling Best Practices
import requests
from requests.exceptions import RequestException, HTTPError, Timeout
import logging
logger = logging.getLogger(__name__)
def query_datadog_with_retry(url, headers, params, max_retries=3):
"""Query Datadog API with retry logic."""
for attempt in range(max_retries):
try:
response = requests.get(
url,
headers=headers,
params=params,
timeout=30
)
response.raise_for_status()
return response.json()
except Timeout:
logger.warning(f"Timeout on attempt {attempt + 1}")
if attempt == max_retries - 1:
raise
except HTTPError as e:
if e.response.status_code == 429: # Rate limit
logger.warning("Rate limited, waiting before retry")
time.sleep(2 ** attempt) # Exponential backoff
else:
raise
except RequestException as e:
logger.error(f"Request failed: {e}")
raise
Rate Limiting Considerations
Datadog API has rate limits. Best practices include:
import time
from functools import wraps
def rate_limit(calls_per_second=10):
"""Decorator to rate limit API calls."""
min_interval = 1.0 / calls_per_second
last_called = [0.0]
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
elapsed = time.time() - last_called[0]
left_to_wait = min_interval - elapsed
if left_to_wait > 0:
time.sleep(left_to_wait)
result = func(*args, **kwargs)
last_called[0] = time.time()
return result
return wrapper
return decorator
@rate_limit(calls_per_second=5)
def query_metrics(query):
"""Rate-limited metrics query."""
# Your query implementation
pass
Pagination Best Practices
Many Datadog API endpoints support pagination:
def get_all_monitors(api_client):
"""Retrieve all monitors with pagination."""
from datadog_api_client.v1.api.monitors_api import MonitorsApi
api = MonitorsApi(api_client)
all_monitors = []
page = 0
page_size = 100
while True:
monitors = api.list_monitors(
page=page,
page_size=page_size
)
if not monitors:
break
all_monitors.extend(monitors)
if len(monitors) < page_size:
break
page += 1
return all_monitors
Complete Example: Comprehensive Monitoring Script
Here’s a complete example that demonstrates both SDK and API approaches:
#!/usr/bin/env python3
"""
Comprehensive Datadog monitoring script.
Retrieves metrics, logs, and product data using both SDK and direct API.
"""
import os
import sys
import time
import logging
from datetime import datetime, timedelta
from datadog_api_client import ApiClient, Configuration
from datadog_api_client.v1.api.metrics_api import MetricsApi
from datadog_api_client.v2.api.logs_api import LogsApi
from datadog_api_client.v1.api.dashboards_api import DashboardsApi
import requests
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class DatadogMonitor:
"""Unified class for Datadog monitoring operations."""
def __init__(self, use_sdk=True):
"""Initialize with SDK or direct API approach."""
self.api_key = os.getenv('DD_API_KEY')
self.app_key = os.getenv('DD_APP_KEY')
self.dd_site = os.getenv('DD_SITE', 'datadoghq.com')
self.use_sdk = use_sdk
if not self.api_key or not self.app_key:
raise ValueError("DD_API_KEY and DD_APP_KEY must be set")
if use_sdk:
self._init_sdk()
else:
self._init_api()
def _init_sdk(self):
"""Initialize SDK client."""
config = Configuration()
config.api_key['apiKeyAuth'] = self.api_key
config.api_key['appKeyAuth'] = self.app_key
config.server_variables['site'] = self.dd_site
self.api_client = ApiClient(config)
def _init_api(self):
"""Initialize direct API settings."""
self.base_url = f"https://api.{self.dd_site}"
self.headers = {
'DD-API-KEY': self.api_key,
'DD-APPLICATION-KEY': self.app_key,
'Content-Type': 'application/json'
}
def get_metrics(self, query, hours=1):
"""Get metrics data."""
if self.use_sdk:
return self._get_metrics_sdk(query, hours)
else:
return self._get_metrics_api(query, hours)
def _get_metrics_sdk(self, query, hours):
"""Get metrics using SDK."""
with self.api_client as api_client:
api = MetricsApi(api_client)
end_time = int(time.time())
start_time = end_time - (hours * 3600)
response = api.query_metrics(
_from=start_time,
to=end_time,
query=query
)
return response
def _get_metrics_api(self, query, hours):
"""Get metrics using direct API."""
end_time = int(time.time())
start_time = end_time - (hours * 3600)
response = requests.get(
f"{self.base_url}/api/v1/query",
headers=self.headers,
params={
'query': query,
'from': start_time,
'to': end_time
}
)
response.raise_for_status()
return response.json()
def generate_report(self):
"""Generate comprehensive monitoring report."""
report = {
'timestamp': datetime.utcnow().isoformat(),
'method': 'SDK' if self.use_sdk else 'Direct API'
}
try:
# Get CPU metrics
logger.info("Fetching CPU metrics...")
cpu_data = self.get_metrics('avg:system.cpu.user{*}')
report['cpu_metrics'] = 'Retrieved successfully'
# Get memory metrics
logger.info("Fetching memory metrics...")
mem_data = self.get_metrics('avg:system.mem.used{*}')
report['memory_metrics'] = 'Retrieved successfully'
logger.info("Report generated successfully")
return report
except Exception as e:
logger.error(f"Error generating report: {e}")
report['error'] = str(e)
return report
def main():
"""Main entry point."""
print("Datadog Monitoring Script")
print("=" * 50)
# Demonstrate SDK approach
print("\n1. Using SDK Approach:")
print("-" * 50)
try:
monitor_sdk = DatadogMonitor(use_sdk=True)
report_sdk = monitor_sdk.generate_report()
print(f"Report: {report_sdk}")
except Exception as e:
print(f"SDK approach failed: {e}")
# Demonstrate Direct API approach
print("\n2. Using Direct API Approach:")
print("-" * 50)
try:
monitor_api = DatadogMonitor(use_sdk=False)
report_api = monitor_api.generate_report()
print(f"Report: {report_api}")
except Exception as e:
print(f"API approach failed: {e}")
if __name__ == "__main__":
main()
Conclusion
Python provides excellent support for integrating with Datadog through multiple approaches:
- Official SDK (
datadog-api-client): Best for production applications requiring type safety and comprehensive API coverage - DogStatsD Client (
datadogpy): Ideal for sending custom metrics from applications - Direct API Calls: Perfect for simple scripts and custom integrations
Choose the approach that best fits your use case:
- For production applications and complex integrations: Use the official SDK
- For simple scripts and one-off tasks: Use direct API calls with the requests library
- For sending metrics from your application: Use the DogStatsD client
Both approaches support the same operations (metrics, logs, dashboards, monitors, SLOs) but with different trade-offs in complexity, flexibility, and maintainability.
Additional Resources
Official Datadog Documentation
- Datadog API Documentation
- Python API Client GitHub
- DogStatsD Documentation
- Datadog Python Library (datadogpy)