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

SDKPurposeBest ForInstallation
datadog-api-clientFull API accessQuerying metrics, logs, dashboards; programmatic managementpip install datadog-api-client
datadogpy (DogStatsD)Metrics submissionSending custom metrics from applicationspip install datadog
APM Tracer (ddtrace)Application tracingDistributed tracing and APMpip install ddtrace
Lambda LayerServerless monitoringAWS Lambda functionsVia 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:

  1. Building Production Applications: SDKs provide better error handling and validation
  2. Type Safety Matters: You want IDE auto-completion and type checking
  3. Complex Integrations: Managing multiple API endpoints and resources
  4. Team Collaboration: Standardized approach for team projects
  5. Long-term Maintenance: SDK updates handle API changes automatically

Use Direct API When:

  1. Simple Scripts: One-off data extraction or reporting scripts
  2. Minimal Dependencies: Don’t want to add external packages
  3. Custom Requirements: Need fine-grained control over requests
  4. Debugging: Understanding exact API behavior
  5. Constrained Environments: Limited package installation options

Comparison Table: SDK vs Direct API

AspectSDK (datadog-api-client)Direct API (requests)
Setup ComplexityMedium (install package, configure)Low (just requests library)
Code VerbosityLow (concise, typed)Medium (more manual work)
Type SafetyYes (type hints, models)No (raw dictionaries)
Error HandlingBuilt-in exceptionsManual with status codes
DocumentationGenerated docs + API docsAPI docs only
IDE SupportExcellent (auto-completion)Basic (no type info)
FlexibilityHigh (covers all endpoints)Very High (complete control)
DependenciesAdditional packageJust requests (often pre-installed)
Learning CurveMedium (SDK-specific patterns)Low (standard HTTP)
Version UpdatesSDK updates with APIMust update manually
Best ForProduction apps, complex integrationsScripts, debugging, simple tools

Best Practices

Security Best Practices

  1. Never Hardcode Credentials: Always use environment variables or secrets management
  2. Rotate Keys Regularly: Implement key rotation policies
  3. Use Application Keys with Minimum Permissions: Create keys with only necessary scopes
  4. 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:

  1. Official SDK (datadog-api-client): Best for production applications requiring type safety and comprehensive API coverage
  2. DogStatsD Client (datadogpy): Ideal for sending custom metrics from applications
  3. 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

API References

Python Resources

Community Resources