# Datadog Python Api > **Source:** --- # Datadog Python API Client - Complete Reference **Source:** https://docs.datadoghq.com/api/latest/?tab=python **Repository:** https://github.com/DataDog/datadog-api-client-python **Official Docs:** https://datadoghq.dev/datadog-api-client-python/ ## Table of Contents 1. [Installation](#installation) 2. [Authentication](#authentication) 3. [Core Components](#core-components) 4. [Configuration](#configuration) 5. [API Operations](#api-operations) 6. [Examples](#examples) 7. [Advanced Features](#advanced-features) --- ## Installation ### Basic Installation Install the Python client library using pip: ```bash pip install datadog-api-client ``` ### With Async Support For asynchronous operations, include the async extra: ```bash pip install datadog-api-client[async] ``` ### Requirements - Python 3.8 or higher - Required dependencies are automatically installed with pip --- ## Authentication ### Environment Variables (Default) The client reads credentials from environment variables by default: ```bash export DD_API_KEY="" export DD_APP_KEY="" ``` ### Manual Configuration Configure credentials directly in your code: ```python from datadog_api_client import ApiClient, Configuration configuration = Configuration() configuration.api_key["apiKeyAuth"] = "" configuration.api_key["appKeyAuth"] = "" ``` ### Server Configuration Switch between Datadog instances (e.g., EU): ```python configuration.server_variables["site"] = "datadoghq.eu" ``` Available sites: - `datadoghq.com` (US, default) - `datadoghq.eu` (EU) --- ## Core Components ### ApiClient (Synchronous) The standard synchronous client for API operations: ```python from datadog_api_client import ApiClient, Configuration configuration = Configuration() with ApiClient(configuration) as api_client: # Use api_client with various API classes pass ``` ### AsyncApiClient (Asynchronous) For async/await patterns with non-blocking operations: ```python from datadog_api_client import AsyncApiClient, Configuration import asyncio configuration = Configuration() async def main(): async with AsyncApiClient(configuration) as api_client: # Use async API operations pass asyncio.run(main()) ``` Requires: `pip install datadog-api-client[async]` ### ThreadedApiClient For thread-based concurrent operations: ```python from datadog_api_client import ThreadedApiClient, Configuration configuration = Configuration() api_client = ThreadedApiClient(configuration) # Returns AsyncResult objects for each API call ``` ### Configuration Object Central configuration management for the API client: ```python from datadog_api_client import Configuration configuration = Configuration() configuration.api_key["apiKeyAuth"] = "" configuration.api_key["appKeyAuth"] = "" configuration.debug = True # Enable request/response logging configuration.compress = False # Disable GZIP compression configuration.server_variables["site"] = "datadoghq.eu" configuration.proxy = "http://proxy.example.com:8080" ``` --- ## Configuration ### Request/Response Compression Control GZIP compression (enabled by default): ```python configuration.compress = False ``` ### Debug Logging Enable detailed request and response logging: ```python configuration.debug = True ``` ### Proxy Configuration Route API requests through a proxy: ```python configuration.proxy = "http://user:password@proxy.example.com:8080" ``` ### Retry Configuration Enable automatic retries for rate-limit errors (429 status): ```python from urllib3.util.retry import Retry from requests.adapters import HTTPAdapter # Simple retry with custom count configuration.max_retries = 5 # Advanced retry policy retry_policy = Retry( total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504] ) ``` ### Unstable/Beta Endpoints Enable experimental API endpoints: ```python configuration.unstable_operations["list_log_indexes"] = True configuration.unstable_operations["create_scenario"] = True ``` --- ## API Operations ### Supported API Versions The library provides access to both v1 and v2 API endpoints: - **v1 API:** Legacy endpoints in `datadog_api_client.v1` - **v2 API:** Current endpoints in `datadog_api_client.v2` ### API Classes (v2) Common API classes available in `datadog_api_client.v2.api`: ```python from datadog_api_client.v2.api import ( APMApi, # Application Performance Monitoring AuditApi, # Audit logs and trails CaseManagementApi, # Case management CloudCostManagementApi, # Cloud cost analysis DashboardsApi, # Dashboard management IncidentsApi, # Incident management LogsApi, # Log ingestion and querying MetricsApi, # Metrics and timeseries MonitorsApi, # Monitor and alerting OpsgenieIntegrationApi, # Opsgenie integration SecurityMonitoringApi, # Security monitoring rules SyntheticsApi, # Synthetic monitoring UsersApi, # User management ) ``` ### Example: Create a Monitor ```python from datadog_api_client.v2.api.monitors_api import MonitorsApi from datadog_api_client.v2.model.monitor import Monitor from datadog_api_client.v2.model.monitor_type import MonitorType body = Monitor( name="Test Monitor", type=MonitorType.METRIC_ALERT, query="avg(last_5m):avg:system.cpu{*} > 0.5", ) with ApiClient(configuration) as api_client: api_instance = MonitorsApi(api_client) response = api_instance.create_monitor(body=body) print(response) ``` --- ## Examples ### Example 1: Get Service List (APM) ```python from datadog_api_client import ApiClient, Configuration from datadog_api_client.v2.api.apm_api import APMApi configuration = Configuration() with ApiClient(configuration) as api_client: api_instance = APMApi(api_client) response = api_instance.get_service_list() print(response) ``` ### Example 2: Create a Monitor ```python from datadog_api_client import ApiClient, Configuration from datadog_api_client.v2.api.monitors_api import MonitorsApi from datadog_api_client.v2.model.monitor import Monitor from datadog_api_client.v2.model.monitor_type import MonitorType configuration = Configuration() body = Monitor( name="My Test Monitor", type=MonitorType.METRIC_ALERT, query="avg(last_5m):avg:system.memory.free{*} > 100000000", message="Test message", ) with ApiClient(configuration) as api_client: api_instance = MonitorsApi(api_client) response = api_instance.create_monitor(body=body) print(response) ``` ### Example 3: Query Logs ```python 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 configuration = Configuration() body = LogsListRequest( filter={ "from": "2024-01-01", "to": "2024-01-31", "query": "status:error", }, page={"limit": 10}, ) with ApiClient(configuration) as api_client: api_instance = LogsApi(api_client) response = api_instance.list_logs(body=body) print(response) ``` ### Example 4: List Dashboards ```python from datadog_api_client import ApiClient, Configuration from datadog_api_client.v2.api.dashboards_api import DashboardsApi configuration = Configuration() with ApiClient(configuration) as api_client: api_instance = DashboardsApi(api_client) response = api_instance.list_dashboards() print(response) ``` ### Example 5: Async Operations ```python from datadog_api_client import AsyncApiClient, Configuration from datadog_api_client.v2.api.apm_api import APMApi import asyncio async def main(): configuration = Configuration() async with AsyncApiClient(configuration) as api_client: api_instance = APMApi(api_client) response = await api_instance.get_service_list() print(response) asyncio.run(main()) ``` --- ## Advanced Features ### Pagination List operations include built-in pagination methods: ```python from datadog_api_client import ApiClient, Configuration from datadog_api_client.v2.api.dashboards_api import DashboardsApi configuration = Configuration() with ApiClient(configuration) as api_client: api_instance = DashboardsApi(api_client) # Retrieve paginated results response = api_instance.list_dashboards(page_size=10) # Iterate through pages for dashboard in response.data: print(dashboard) ``` ### Rate Limit Handling The library automatically handles rate limits (429 responses) when retry is configured: ```python configuration.max_retries = 5 # Automatically retry on 429 ``` The Retry-After header is respected, and backoff is applied automatically. ### Error Handling Handle API errors gracefully: ```python from datadog_api_client.exceptions import ApiException try: response = api_instance.create_monitor(body=body) except ApiException as e: print(f"Error: {e.status} - {e.reason}") print(f"Response body: {e.body}") ``` ### Context Manager Pattern Use context managers for proper resource cleanup: ```python with ApiClient(configuration) as api_client: api_instance = MonitorsApi(api_client) response = api_instance.list_monitors() # Client automatically closed after the with block ``` ### Custom Headers Add custom headers to API requests: ```python configuration.api_key_prefix = {"X-Custom-Header": "value"} ``` ### Timeouts Configure request timeouts: ```python import requests configuration.timeout = 10 # seconds ``` --- ## Common Use Cases ### User Management ```python from datadog_api_client.v2.api.users_api import UsersApi with ApiClient(configuration) as api_client: api_instance = UsersApi(api_client) users = api_instance.list_users() for user in users.data: print(f"{user.attributes['name']} ({user.attributes['email']})") ``` ### Incident Management ```python from datadog_api_client.v2.api.incidents_api import IncidentsApi with ApiClient(configuration) as api_client: api_instance = IncidentsApi(api_client) incidents = api_instance.list_incidents() for incident in incidents.data: print(f"Incident: {incident.attributes['title']}") ``` ### Security Monitoring ```python from datadog_api_client.v2.api.security_monitoring_api import SecurityMonitoringApi with ApiClient(configuration) as api_client: api_instance = SecurityMonitoringApi(api_client) rules = api_instance.list_rules() for rule in rules.data: print(f"Rule: {rule.attributes['name']}") ``` --- ## Troubleshooting ### Debug Mode Enable debug logging to see full request/response details: ```python configuration.debug = True ``` ### Common Issues **Issue:** "Unauthorized" error (401) - **Solution:** Verify DD_API_KEY and DD_APP_KEY are set correctly - Check that API key has appropriate permissions **Issue:** "Not Found" error (404) - **Solution:** Verify the correct site is configured (EU vs US) - Check that the resource exists **Issue:** "Too Many Requests" error (429) - **Solution:** Enable retry mechanism with `configuration.max_retries` - Implement exponential backoff ### Environment Variables Verify environment variables are set: ```bash echo $DD_API_KEY echo $DD_APP_KEY ``` --- ## Resources - **Official Repository:** https://github.com/DataDog/datadog-api-client-python - **API Documentation:** https://datadoghq.dev/datadog-api-client-python/ - **Datadog Docs:** https://docs.datadoghq.com/ - **API Reference:** https://docs.datadoghq.com/api/latest/?tab=python - **Issue Tracker:** https://github.com/DataDog/datadog-api-client-python/issues --- ## Support - **Datadog Support:** https://docs.datadoghq.com/help - **Community:** Slack, GitHub Discussions - **Contributing:** See CONTRIBUTING.md in the repository --- # Datadog Python API Client - Best Practices & Troubleshooting **Guidelines for reliable and efficient Datadog API usage** ## Table of Contents 1. [Authentication Best Practices](#authentication-best-practices) 2. [Error Handling Strategies](#error-handling-strategies) 3. [Performance Optimization](#performance-optimization) 4. [Rate Limiting](#rate-limiting) 5. [Logging and Debugging](#logging-and-debugging) 6. [Common Pitfalls](#common-pitfalls) 7. [Testing](#testing) 8. [Troubleshooting](#troubleshooting) --- ## Authentication Best Practices ### 1. Use Environment Variables Never hardcode credentials in your application: ```python # Bad - DO NOT DO THIS configuration.api_key["apiKeyAuth"] = "your_api_key_here" # Good - Use environment variables import os from datadog_api_client import Configuration configuration = Configuration() # Automatically reads from DD_API_KEY and DD_APP_KEY ``` ### 2. Secure Credential Storage For production applications, use secure credential management: ```python # Using python-dotenv for local development from dotenv import load_dotenv import os load_dotenv() # Load from .env file # Using AWS Secrets Manager for AWS deployments import boto3 import json def get_datadog_credentials(): """Retrieve Datadog credentials from AWS Secrets Manager""" secrets_client = boto3.client('secretsmanager') secret = secrets_client.get_secret_value(SecretId='datadog/api-keys') credentials = json.loads(secret['SecretString']) return credentials['api_key'], credentials['app_key'] # For Azure from azure.identity import DefaultAzureCredential from azure.keyvault.secrets import SecretClient credential = DefaultAzureCredential() client = SecretClient(vault_url="https://your-keyvault.vault.azure.net", credential=credential) api_key = client.get_secret("datadog-api-key").value ``` ### 3. Validate Credentials on Startup ```python from datadog_api_client import ApiClient, Configuration from datadog_api_client.v2.api.users_api import UsersApi import sys def validate_credentials(): """Validate Datadog credentials""" configuration = Configuration() try: with ApiClient(configuration) as api_client: users_api = UsersApi(api_client) # Simple API call to verify authentication users_api.list_users(page_size=1) print("✓ Credentials validated successfully") return True except Exception as e: print(f"✗ Authentication failed: {e}") sys.exit(1) if __name__ == "__main__": validate_credentials() # Continue with main application ``` ### 4. Rotate API Keys Regularly ```python # Implement key rotation strategy """ 1. Generate new API key in Datadog UI 2. Update application to use new key 3. Wait for key propagation (usually immediate) 4. Revoke old key in UI 5. Monitor for any authentication errors """ ``` --- ## Error Handling Strategies ### 1. Comprehensive Error Handling ```python from datadog_api_client.exceptions import ApiException import json import logging logger = logging.getLogger(__name__) def safe_api_call(api_func, *args, **kwargs): """Wrapper for safe API calls with comprehensive error handling""" try: return api_func(*args, **kwargs) except ApiException as e: logger.error(f"Datadog API Error: {e.status} {e.reason}") # Parse error body if available try: error_data = json.loads(e.body) logger.error(f"Error details: {error_data}") except: logger.error(f"Error body: {e.body}") # Handle specific status codes if e.status == 401: logger.error("Authentication failed - check credentials") elif e.status == 403: logger.error("Insufficient permissions") elif e.status == 404: logger.error("Resource not found") elif e.status == 429: logger.error("Rate limit exceeded") elif e.status >= 500: logger.error("Server error - retry later") raise except ConnectionError as e: logger.error(f"Connection error: {e}") raise except Exception as e: logger.error(f"Unexpected error: {type(e).__name__}: {e}") raise ``` ### 2. Retry with Exponential Backoff ```python import time from functools import wraps from datadog_api_client.exceptions import ApiException def retry_with_backoff(max_retries=3, base_delay=1, max_delay=60): """Decorator for retry with exponential backoff""" def decorator(func): @wraps(func) def wrapper(*args, **kwargs): for attempt in range(max_retries): try: return func(*args, **kwargs) except ApiException as e: # Only retry on transient errors if e.status not in [429, 500, 502, 503, 504]: raise if attempt == max_retries - 1: raise # Calculate backoff with jitter delay = min(base_delay * (2 ** attempt), max_delay) import random delay += random.uniform(0, delay * 0.1) print(f"Attempt {attempt + 1} failed, retrying in {delay:.1f}s...") time.sleep(delay) return wrapper return decorator # Usage @retry_with_backoff(max_retries=3) def create_monitor_with_retry(api, monitor): return api.create_monitor(body=monitor) ``` --- ## Performance Optimization ### 1. Use Connection Pooling ```python from datadog_api_client import ApiClient, Configuration # Reuse a single ApiClient instance for multiple operations configuration = Configuration() with ApiClient(configuration) as api_client: # All API operations use the same connection monitors_api = MonitorsApi(api_client) dashboards_api = DashboardsApi(api_client) monitors = monitors_api.list_monitors() dashboards = dashboards_api.list_dashboards() # Single connection, multiple API calls ``` ### 2. Disable Unnecessary Compression For local testing or low-bandwidth scenarios: ```python configuration = Configuration() configuration.compress = False # Faster for small payloads ``` ### 3. Batch Operations ```python def batch_create_monitors(monitors_data, batch_size=10): """Create multiple monitors efficiently""" from datadog_api_client.v2.api.monitors_api import MonitorsApi from datadog_api_client.v2.model.monitor import Monitor from datadog_api_client.v2.model.monitor_type import MonitorType results = [] with ApiClient(configuration) as api_client: api = MonitorsApi(api_client) for i, data in enumerate(monitors_data): monitor = Monitor( name=data["name"], type=MonitorType.METRIC_ALERT, query=data["query"], message=data["message"], ) try: result = api.create_monitor(body=monitor) results.append({"success": True, "id": result.id}) except Exception as e: results.append({"success": False, "error": str(e)}) # Log progress if (i + 1) % batch_size == 0: print(f"Processed {i + 1}/{len(monitors_data)}") return results ``` ### 4. Lazy Load Data ```python from datadog_api_client.v2.api.monitors_api import MonitorsApi def get_monitors_lazy(api): """Generator for memory-efficient pagination""" page = 1 while True: monitors = api.list_monitors(page_number=page, page_size=100) if not monitors.data: break for monitor in monitors.data: yield monitor page += 1 # Usage with ApiClient(configuration) as api_client: api = MonitorsApi(api_client) for monitor in get_monitors_lazy(api): print(monitor.name) ``` --- ## Rate Limiting ### 1. Built-in Rate Limit Handling ```python from datadog_api_client import Configuration configuration = Configuration() # Enable automatic retry on rate limit (429) configuration.max_retries = 5 # With custom retry policy from urllib3.util.retry import Retry retry_policy = Retry( total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["GET", "POST", "PUT", "DELETE"] ) ``` ### 2. Monitor Rate Limit Headers ```python from datadog_api_client import ApiClient, Configuration from datadog_api_client.v2.api.monitors_api import MonitorsApi configuration = Configuration() with ApiClient(configuration) as api_client: api = MonitorsApi(api_client) response = api.list_monitors() # Check rate limit headers if available if hasattr(api_client, 'last_response'): headers = api_client.last_response.headers remaining = headers.get('X-RateLimit-Remaining') reset = headers.get('X-RateLimit-Reset') print(f"Rate limit remaining: {remaining}") print(f"Rate limit reset: {reset}") ``` ### 3. Implement Request Throttling ```python import time from threading import Lock class ThrottledApiClient: """API client with request throttling""" def __init__(self, requests_per_second=10): self.requests_per_second = requests_per_second self.min_request_interval = 1.0 / requests_per_second self.last_request_time = 0 self.lock = Lock() def wait_if_needed(self): """Wait if necessary to maintain rate limit""" with self.lock: elapsed = time.time() - self.last_request_time if elapsed < self.min_request_interval: time.sleep(self.min_request_interval - elapsed) self.last_request_time = time.time() def call(self, api_func, *args, **kwargs): """Call API function with throttling""" self.wait_if_needed() return api_func(*args, **kwargs) # Usage throttler = ThrottledApiClient(requests_per_second=5) with ApiClient(configuration) as api_client: api = MonitorsApi(api_client) monitors = throttler.call(api.list_monitors) ``` --- ## Logging and Debugging ### 1. Enable Debug Mode ```python import logging # Enable HTTP logging logging.basicConfig() logging.getLogger("urllib3").setLevel(logging.DEBUG) # Enable Datadog API client logging configuration = Configuration() configuration.debug = True ``` ### 2. Structured Logging ```python import logging import json from datetime import datetime class DatadogApiLogger: """Structured logging for Datadog API operations""" def __init__(self, logger_name="datadog_api"): self.logger = logging.getLogger(logger_name) def log_api_call(self, method, endpoint, status, duration, error=None): """Log API call details""" log_data = { "timestamp": datetime.utcnow().isoformat(), "method": method, "endpoint": endpoint, "status": status, "duration_ms": duration, } if error: log_data["error"] = str(error) self.logger.error(json.dumps(log_data)) else: self.logger.info(json.dumps(log_data)) # Usage logger = DatadogApiLogger() import time start = time.time() try: monitors = api.list_monitors() duration = (time.time() - start) * 1000 logger.log_api_call("GET", "/api/v2/monitors", 200, duration) except Exception as e: duration = (time.time() - start) * 1000 logger.log_api_call("GET", "/api/v2/monitors", None, duration, error=e) ``` --- ## Common Pitfalls ### 1. Not Closing Connections Bad: ```python # BAD: Connection left open api_client = ApiClient(configuration) monitors = api_client.monitors_api.list_monitors() # api_client never closed! ``` Good: ```python # GOOD: Automatic cleanup with context manager with ApiClient(configuration) as api_client: monitors_api = MonitorsApi(api_client) monitors = monitors_api.list_monitors() # Connection closed automatically ``` ### 2. Reusing Stale Configuration ```python # BAD: Creating new ApiClient each time for i in range(100): with ApiClient(configuration) as api_client: # Creates 100 connections pass # GOOD: Reuse connection with ApiClient(configuration) as api_client: for i in range(100): # Single connection for 100 operations pass ``` ### 3. Missing Error Details ```python # BAD: Swallowing error details try: monitors = api.list_monitors() except: print("Failed") # GOOD: Preserve error information import traceback try: monitors = api.list_monitors() except Exception as e: logger.error(f"Failed to list monitors: {e}", exc_info=True) raise ``` ### 4. Hardcoding IDs ```python # BAD: Hardcoded IDs monitor_id = 123456 api.delete_monitor(monitor_id=monitor_id) # GOOD: Look up IDs dynamically monitors = api.list_monitors(name="My Monitor") if monitors.data: for monitor in monitors.data: api.delete_monitor(monitor_id=monitor.id) ``` --- ## Testing ### 1. Mock API Calls ```python from unittest.mock import Mock, patch from datadog_api_client.v2.api.monitors_api import MonitorsApi def test_create_monitor(): """Test monitor creation with mocking""" mock_response = Mock() mock_response.id = 123 mock_response.name = "Test Monitor" with patch.object(MonitorsApi, 'create_monitor', return_value=mock_response): api = MonitorsApi(Mock()) result = api.create_monitor(body=Mock()) assert result.id == 123 assert result.name == "Test Monitor" ``` ### 2. Test with Fixtures ```python import pytest from datadog_api_client import ApiClient, Configuration @pytest.fixture def api_client(): """Fixture for API client""" configuration = Configuration() with ApiClient(configuration) as client: yield client def test_list_monitors(api_client): """Test listing monitors""" from datadog_api_client.v2.api.monitors_api import MonitorsApi api = MonitorsApi(api_client) monitors = api.list_monitors() assert len(monitors.data) >= 0 ``` --- ## Troubleshooting ### Problem: "Unauthorized" (401) ``` Datadog API error: 401 UNAUTHORIZED ``` **Solutions:** 1. Verify credentials are set: ```bash echo $DD_API_KEY echo $DD_APP_KEY ``` 2. Verify credentials in configuration: ```python print(configuration.api_key) ``` 3. Check API key permissions in Datadog UI 4. Regenerate API key if suspected compromise ### Problem: "Rate Limited" (429) ``` Datadog API error: 429 TOO_MANY_REQUESTS ``` **Solutions:** ```python # Enable automatic retry configuration.max_retries = 5 # Or implement manual backoff import time time.sleep(30) # Wait 30 seconds ``` ### Problem: "Not Found" (404) ``` Datadog API error: 404 NOT_FOUND ``` **Solutions:** 1. Verify resource exists 2. Check correct site: ```python configuration.server_variables["site"] = "datadoghq.com" # or datadoghq.eu ``` 3. Verify ID format: ```python print(f"Monitor ID: {monitor.id} (type: {type(monitor.id)})") ``` ### Problem: SSL Certificate Error ``` ssl.SSLError: certificate verify failed ``` **Solutions (dev only, never in production):** ```python import ssl import urllib3 # Disable certificate verification (INSECURE!) urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) # Via environment variable import os os.environ['CURL_CA_BUNDLE'] = '' ``` **Proper solution:** ```bash # Update certificate bundle pip install --upgrade certifi # Or use specific CA bundle export REQUESTS_CA_BUNDLE=/path/to/ca-bundle.crt ``` --- ## Resources - **Official Docs:** https://docs.datadoghq.com/ - **API Reference:** https://docs.datadoghq.com/api/latest/?tab=python - **GitHub Issues:** https://github.com/DataDog/datadog-api-client-python/issues - **Datadog Support:** https://docs.datadoghq.com/help/ --- # Datadog Python API Client - Code Examples **Practical examples for common Datadog operations** ## Table of Contents 1. [Monitor Operations](#monitor-operations) 2. [Dashboard Operations](#dashboard-operations) 3. [Log Operations](#log-operations) 4. [APM Operations](#apm-operations) 5. [User Management](#user-management) 6. [Incident Management](#incident-management) 7. [Advanced Patterns](#advanced-patterns) 8. [Error Handling](#error-handling) --- ## Monitor Operations ### Create a Metric Alert Monitor ```python from datadog_api_client import ApiClient, Configuration from datadog_api_client.v2.api.monitors_api import MonitorsApi from datadog_api_client.v2.model.monitor import Monitor from datadog_api_client.v2.model.monitor_type import MonitorType configuration = Configuration() monitor = Monitor( name="High CPU Usage Alert", type=MonitorType.METRIC_ALERT, query="avg(last_5m):avg:system.cpu{*} > 0.8", message=""" CPU usage is too high! Value: {{value}} Host: {{host.name}} """, tags=["production", "cpu"], enabled=True, priority=1, no_data_timeframe=10, # Alert if no data for 10 minutes ) with ApiClient(configuration) as api_client: monitors_api = MonitorsApi(api_client) response = monitors_api.create_monitor(body=monitor) print(f"Created monitor with ID: {response.id}") ``` ### Create a Log Alert Monitor ```python from datadog_api_client.v2.model.monitor_type import MonitorType monitor = Monitor( name="Critical Errors Alert", type=MonitorType.LOG_ALERT, query='status:error source:production_app', message="Critical errors detected in production", tags=["production", "logs"], ) with ApiClient(configuration) as api_client: monitors_api = MonitorsApi(api_client) response = monitors_api.create_monitor(body=monitor) ``` ### List All Monitors ```python with ApiClient(configuration) as api_client: monitors_api = MonitorsApi(api_client) monitors = monitors_api.list_monitors() for monitor in monitors.data: print(f"Monitor: {monitor.name}") print(f" ID: {monitor.id}") print(f" Type: {monitor.type}") print(f" State: {monitor.state}") print() ``` ### Get a Specific Monitor ```python monitor_id = 12345 with ApiClient(configuration) as api_client: monitors_api = MonitorsApi(api_client) monitor = monitors_api.get_monitor(monitor_id=monitor_id) print(f"Name: {monitor.name}") print(f"Query: {monitor.query}") print(f"Message: {monitor.message}") ``` ### Update a Monitor ```python monitor_id = 12345 with ApiClient(configuration) as api_client: monitors_api = MonitorsApi(api_client) # Get existing monitor monitor = monitors_api.get_monitor(monitor_id=monitor_id) # Update properties monitor.message = "Updated: " + monitor.message monitor.tags.append("updated") # Send update updated = monitors_api.update_monitor( monitor_id=monitor_id, body=monitor ) print(f"Updated monitor: {updated.name}") ``` ### Delete a Monitor ```python monitor_id = 12345 with ApiClient(configuration) as api_client: monitors_api = MonitorsApi(api_client) monitors_api.delete_monitor(monitor_id=monitor_id) print(f"Deleted monitor {monitor_id}") ``` ### Mute a Monitor ```python from datadog_api_client.v2.model.monitor_update_request import MonitorUpdateRequest monitor_id = 12345 with ApiClient(configuration) as api_client: monitors_api = MonitorsApi(api_client) # Mute all instances monitors_api.update_monitor_state( monitor_id=monitor_id, body=MonitorUpdateRequest( action="mute" ) ) ``` ### Search Monitors by Tags ```python with ApiClient(configuration) as api_client: monitors_api = MonitorsApi(api_client) # List monitors with specific tag monitors = monitors_api.list_monitors( tags=["production"] ) print(f"Found {len(monitors.data)} production monitors") ``` --- ## Dashboard Operations ### Create a Dashboard with Widgets ```python from datadog_api_client.v2.api.dashboards_api import DashboardsApi from datadog_api_client.v2.model.dashboard import Dashboard from datadog_api_client.v2.model.dashboard_type import DashboardType from datadog_api_client.v2.model.widget import Widget from datadog_api_client.v2.model.widget_definition import WidgetDefinition from datadog_api_client.v2.model.widget_type import WidgetType dashboard = Dashboard( title="System Performance", description="Key system metrics dashboard", type=DashboardType.TIMEBOARD, tags=["system", "performance"], widgets=[ Widget( definition=WidgetDefinition( type=WidgetType.TIMESERIES, requests=[{ "queries": [{ "query": "avg:system.cpu{*}" }] }], title="CPU Usage" ) ), Widget( definition=WidgetDefinition( type=WidgetType.QUERY_VALUE, requests=[{ "queries": [{ "query": "avg:system.mem.pct_usable{*}" }] }], title="Memory Available %" ) ) ] ) with ApiClient(configuration) as api_client: dashboards_api = DashboardsApi(api_client) response = dashboards_api.create_dashboard(body=dashboard) print(f"Created dashboard: {response.id}") ``` ### List Dashboards ```python with ApiClient(configuration) as api_client: dashboards_api = DashboardsApi(api_client) dashboards = dashboards_api.list_dashboards() for dashboard in dashboards.data: print(f"Dashboard: {dashboard.title}") print(f" ID: {dashboard.id}") print(f" Type: {dashboard.type}") ``` ### Get a Dashboard ```python dashboard_id = "abc-123-def" with ApiClient(configuration) as api_client: dashboards_api = DashboardsApi(api_client) dashboard = dashboards_api.get_dashboard(dashboard_id=dashboard_id) print(f"Title: {dashboard.title}") print(f"Widgets: {len(dashboard.widgets)}") ``` ### Update a Dashboard ```python dashboard_id = "abc-123-def" with ApiClient(configuration) as api_client: dashboards_api = DashboardsApi(api_client) # Get dashboard dashboard = dashboards_api.get_dashboard(dashboard_id=dashboard_id) # Modify dashboard.title = "Updated: " + dashboard.title # Update updated = dashboards_api.update_dashboard( dashboard_id=dashboard_id, body=dashboard ) ``` ### Delete a Dashboard ```python dashboard_id = "abc-123-def" with ApiClient(configuration) as api_client: dashboards_api = DashboardsApi(api_client) dashboards_api.delete_dashboard(dashboard_id=dashboard_id) ``` --- ## Log Operations ### Query Logs ```python from datadog_api_client.v2.api.logs_api import LogsApi from datadog_api_client.v2.model.logs_list_request import LogsListRequest request = LogsListRequest( filter={ "from": "2024-01-01T00:00:00Z", "to": "2024-01-31T23:59:59Z", "query": "status:error source:api", }, page={ "limit": 100, }, ) with ApiClient(configuration) as api_client: logs_api = LogsApi(api_client) response = logs_api.list_logs(body=request) for log in response.data: print(f"[{log.attributes['status']}] {log.attributes['message']}") ``` ### Query Logs with Aggregation ```python from datadog_api_client.v2.model.logs_aggregate_request import LogsAggregateRequest from datadog_api_client.v2.model.logs_metric_compute import LogsMetricCompute from datadog_api_client.v2.model.logs_group_by import LogsGroupBy request = LogsAggregateRequest( filter={ "from": "2024-01-01T00:00:00Z", "to": "2024-01-31T23:59:59Z", "query": "source:api", }, compute=[ LogsMetricCompute( aggregation="count", ) ], group_by=[ LogsGroupBy(path="@status", limit=10) ], ) with ApiClient(configuration) as api_client: logs_api = LogsApi(api_client) response = logs_api.aggregate_logs(body=request) for bucket in response.data.buckets: print(f"Status: {bucket.key}") print(f" Count: {bucket.computes[0].value}") ``` ### Get Log Details ```python log_id = "abc123" with ApiClient(configuration) as api_client: logs_api = LogsApi(api_client) log = logs_api.get_log(log_id=log_id) print(f"Message: {log.attributes['message']}") print(f"Status: {log.attributes['status']}") print(f"Service: {log.attributes.get('service', 'N/A')}") ``` --- ## APM Operations ### Get Service List ```python from datadog_api_client.v2.api.apm_api import APMApi with ApiClient(configuration) as api_client: apm_api = APMApi(api_client) response = apm_api.get_service_list() print(response) ``` ### Get Service Statistics ```python service_name = "my-api" with ApiClient(configuration) as api_client: apm_api = APMApi(api_client) stats = apm_api.get_service_stats(service_name=service_name) print(f"Service: {service_name}") print(f" Requests/sec: {stats.stats.throughput}") print(f" Latency (ms): {stats.stats.latency}") print(f" Error rate: {stats.stats.errors}%") ``` --- ## User Management ### List Users ```python from datadog_api_client.v2.api.users_api import UsersApi with ApiClient(configuration) as api_client: users_api = UsersApi(api_client) users = users_api.list_users() for user in users.data: print(f"User: {user.attributes['name']}") print(f" Email: {user.attributes['email']}") print(f" Status: {user.attributes['status']}") ``` ### Create a User ```python from datadog_api_client.v2.model.user_create_request import UserCreateRequest user_request = UserCreateRequest( data={ "type": "users", "attributes": { "name": "John Doe", "email": "john.doe@example.com", } } ) with ApiClient(configuration) as api_client: users_api = UsersApi(api_client) response = users_api.create_user(body=user_request) print(f"Created user: {response.data.id}") ``` ### Update User Status ```python user_id = "user123" from datadog_api_client.v2.model.user_update_request import UserUpdateRequest update_request = UserUpdateRequest( data={ "type": "users", "id": user_id, "attributes": { "status": "disabled" # or "active" } } ) with ApiClient(configuration) as api_client: users_api = UsersApi(api_client) users_api.update_user(user_id=user_id, body=update_request) ``` --- ## Incident Management ### List Incidents ```python from datadog_api_client.v2.api.incidents_api import IncidentsApi with ApiClient(configuration) as api_client: incidents_api = IncidentsApi(api_client) incidents = incidents_api.list_incidents() for incident in incidents.data: print(f"Incident: {incident.attributes['title']}") print(f" Status: {incident.attributes['status']}") print(f" Severity: {incident.attributes['severity']}") ``` ### Create an Incident ```python from datadog_api_client.v2.model.incident_create_request import IncidentCreateRequest incident_request = IncidentCreateRequest( data={ "type": "incidents", "attributes": { "title": "Database connection pool exhausted", "status": "active", "severity": "critical", } } ) with ApiClient(configuration) as api_client: incidents_api = IncidentsApi(api_client) response = incidents_api.create_incident(body=incident_request) print(f"Created incident: {response.data.id}") ``` ### Update Incident ```python incident_id = "incident123" from datadog_api_client.v2.model.incident_update_request import IncidentUpdateRequest update_request = IncidentUpdateRequest( data={ "type": "incidents", "id": incident_id, "attributes": { "status": "resolved", "customer_impact_scope": "Database is now stable", } } ) with ApiClient(configuration) as api_client: incidents_api = IncidentsApi(api_client) incidents_api.update_incident( incident_id=incident_id, body=update_request ) ``` --- ## Advanced Patterns ### Batch Create Monitors ```python monitors_config = [ { "name": "CPU Alert", "query": "avg(last_5m):avg:system.cpu{*} > 0.8", }, { "name": "Memory Alert", "query": "avg(last_5m):avg:system.mem.pct_usable{*} < 0.1", }, { "name": "Disk Alert", "query": "avg(last_5m):avg:system.disk.free{*} < 1000000000", }, ] with ApiClient(configuration) as api_client: monitors_api = MonitorsApi(api_client) for config in monitors_config: monitor = Monitor( name=config["name"], type=MonitorType.METRIC_ALERT, query=config["query"], message=f"Alert: {config['name']}", tags=["batch-created"], ) response = monitors_api.create_monitor(body=monitor) print(f"Created: {response.id}") ``` ### Async Operations ```python import asyncio from datadog_api_client import AsyncApiClient, Configuration from datadog_api_client.v2.api.monitors_api import MonitorsApi async def get_monitors(): configuration = Configuration() async with AsyncApiClient(configuration) as api_client: monitors_api = MonitorsApi(api_client) monitors = await monitors_api.list_monitors() return monitors # Run monitors = asyncio.run(get_monitors()) print(f"Found {len(monitors.data)} monitors") ``` ### Implement Retry Logic ```python import time from datadog_api_client.exceptions import ApiException def create_monitor_with_retry(monitors_api, monitor, max_retries=3): """Create monitor with automatic retry on transient errors""" for attempt in range(max_retries): try: return monitors_api.create_monitor(body=monitor) except ApiException as e: if e.status in [429, 500, 502, 503]: # Transient errors wait_time = 2 ** attempt # Exponential backoff print(f"Attempt {attempt + 1} failed, retrying in {wait_time}s...") time.sleep(wait_time) else: raise # Non-transient error, fail immediately with ApiClient(configuration) as api_client: monitors_api = MonitorsApi(api_client) response = create_monitor_with_retry(monitors_api, monitor) ``` --- ## Error Handling ### Handle All Error Types ```python from datadog_api_client.exceptions import ApiException import json with ApiClient(configuration) as api_client: monitors_api = MonitorsApi(api_client) try: response = monitors_api.create_monitor(body=monitor) except ApiException as e: print(f"HTTP Status: {e.status}") print(f"Reason: {e.reason}") # Parse error response try: error_data = json.loads(e.body) print(f"Error Details: {error_data}") except: print(f"Raw Response: {e.body}") except ConnectionError as e: print(f"Network error: {e}") except Exception as e: print(f"Unexpected error: {e}") ``` ### Validate Input Before Sending ```python def validate_monitor(monitor): """Validate monitor before sending to API""" errors = [] if not monitor.name or len(monitor.name) < 3: errors.append("Monitor name must be at least 3 characters") if not monitor.query: errors.append("Monitor query is required") if not monitor.message: errors.append("Monitor message is required") if monitor.type not in MonitorType: errors.append(f"Invalid monitor type: {monitor.type}") if errors: raise ValueError("; ".join(errors)) return True # Use try: validate_monitor(monitor) response = monitors_api.create_monitor(body=monitor) except ValueError as e: print(f"Validation error: {e}") ``` --- ## Full Working Example Complete end-to-end example: ```python #!/usr/bin/env python3 """ Complete Datadog API example: Create monitor, list dashboards, query logs """ import os from datadog_api_client import ApiClient, Configuration from datadog_api_client.v2.api.monitors_api import MonitorsApi from datadog_api_client.v2.api.dashboards_api import DashboardsApi from datadog_api_client.v2.api.logs_api import LogsApi from datadog_api_client.v2.model.monitor import Monitor from datadog_api_client.v2.model.monitor_type import MonitorType def main(): # Configuration configuration = Configuration() # Optional: Check environment if not os.environ.get("DD_API_KEY"): print("Error: DD_API_KEY not set") return with ApiClient(configuration) as api_client: # Create a monitor print("Creating monitor...") monitors_api = MonitorsApi(api_client) monitor = Monitor( name="Example Monitor", type=MonitorType.METRIC_ALERT, query="avg(last_5m):avg:system.cpu{*} > 0.8", message="CPU is high", ) monitor_response = monitors_api.create_monitor(body=monitor) print(f"✓ Created monitor {monitor_response.id}") # List dashboards print("\nListing dashboards...") dashboards_api = DashboardsApi(api_client) dashboards = dashboards_api.list_dashboards() print(f"✓ Found {len(dashboards.data)} dashboards") for dashboard in dashboards.data[:5]: print(f" - {dashboard.title}") # List monitors print("\nListing monitors...") monitors = monitors_api.list_monitors() print(f"✓ Found {len(monitors.data)} monitors") if __name__ == "__main__": main() ``` --- ## Resources - **GitHub Repository:** https://github.com/DataDog/datadog-api-client-python - **API Documentation:** https://docs.datadoghq.com/api/latest/?tab=python - **Python Client Docs:** https://datadoghq.dev/datadog-api-client-python/ --- # Datadog Python API Client - Getting Started Guide **Quick Start Guide for the datadog-api-client-python library** ## Prerequisites - Python 3.8 or higher - pip (Python package manager) - Datadog account with API and Application keys ## Step 1: Install the Library ```bash # Basic installation pip install datadog-api-client # With async support pip install datadog-api-client[async] ``` Verify the installation: ```bash python -c "from datadog_api_client import ApiClient; print('Installation successful')" ``` ## Step 2: Get Your API Keys 1. Navigate to **Organization Settings** in your Datadog account 2. Select **API Keys** from the left sidebar 3. Click **New Key** to generate an API key 4. Copy the key value 5. Select **Application Keys** 6. Click **New Key** to generate an application key 7. Copy the application key value **Security Note:** Never commit API keys to version control. Use environment variables instead. ## Step 3: Set Environment Variables ```bash # Linux/macOS export DD_API_KEY="your_api_key_here" export DD_APP_KEY="your_application_key_here" # Windows (PowerShell) $env:DD_API_KEY="your_api_key_here" $env:DD_APP_KEY="your_application_key_here" # Windows (Command Prompt) set DD_API_KEY=your_api_key_here set DD_APP_KEY=your_application_key_here ``` Or add to your `.bashrc`, `.zshrc`, or Windows environment variables for persistence. ## Step 4: Verify Authentication Create a test script `test_auth.py`: ```python from datadog_api_client import ApiClient, Configuration configuration = Configuration() with ApiClient(configuration) as api_client: print("Connection successful!") print(f"Using configuration: {configuration}") ``` Run the script: ```bash python test_auth.py ``` ## Step 5: Your First API Call Create a script to list monitors: ```python from datadog_api_client import ApiClient, Configuration from datadog_api_client.v2.api.monitors_api import MonitorsApi configuration = Configuration() with ApiClient(configuration) as api_client: monitors_api = MonitorsApi(api_client) monitors = monitors_api.list_monitors() print(f"Total monitors: {len(monitors.data)}") for monitor in monitors.data: print(f"- {monitor.name} (ID: {monitor.id})") ``` ## Common First-Time Tasks ### Task 1: Create a Simple Monitor ```python from datadog_api_client import ApiClient, Configuration from datadog_api_client.v2.api.monitors_api import MonitorsApi from datadog_api_client.v2.model.monitor import Monitor from datadog_api_client.v2.model.monitor_type import MonitorType configuration = Configuration() monitor_body = Monitor( name="My First Monitor", type=MonitorType.METRIC_ALERT, query="avg(last_5m):avg:system.cpu{*} > 0.8", message="CPU usage is high: {{value}}", ) with ApiClient(configuration) as api_client: monitors_api = MonitorsApi(api_client) response = monitors_api.create_monitor(body=monitor_body) print(f"Created monitor: {response.id}") ``` ### Task 2: List Dashboards ```python from datadog_api_client import ApiClient, Configuration from datadog_api_client.v2.api.dashboards_api import DashboardsApi configuration = Configuration() with ApiClient(configuration) as api_client: dashboards_api = DashboardsApi(api_client) dashboards = dashboards_api.list_dashboards() for dashboard in dashboards.data: print(f"Dashboard: {dashboard.title}") ``` ### Task 3: Get APM Service Data ```python from datadog_api_client import ApiClient, Configuration from datadog_api_client.v2.api.apm_api import APMApi configuration = Configuration() with ApiClient(configuration) as api_client: apm_api = APMApi(api_client) services = apm_api.get_service_list() print(response) ``` ### Task 4: Query Logs ```python 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 configuration = Configuration() body = LogsListRequest( filter={ "from": "2024-01-01T00:00:00Z", "to": "2024-01-31T23:59:59Z", "query": "status:error source:my_app", }, page={"limit": 100}, ) with ApiClient(configuration) as api_client: logs_api = LogsApi(api_client) response = logs_api.list_logs(body=body) print(f"Found {len(response.data)} logs") ``` ## Configuration Options The `Configuration` object allows you to customize behavior: ```python from datadog_api_client import Configuration configuration = Configuration() # Authentication configuration.api_key["apiKeyAuth"] = "your_api_key" configuration.api_key["appKeyAuth"] = "your_app_key" # Server (region) configuration.server_variables["site"] = "datadoghq.eu" # For EU # Logging & Debug configuration.debug = True # Print request/response details # Performance configuration.compress = False # Disable GZIP compression # Retries configuration.max_retries = 3 # Retry on 429 (rate limit) # Proxy configuration.proxy = "http://proxy.example.com:8080" ``` ## Regions The client supports multiple Datadog regions: ```python # US (default) configuration.server_variables["site"] = "datadoghq.com" # EU configuration.server_variables["site"] = "datadoghq.eu" # US3 configuration.server_variables["site"] = "us3.datadoghq.com" # US1-FED configuration.server_variables["site"] = "us1-fed.datadoghq.com" # AP1 configuration.server_variables["site"] = "ap1.datadoghq.com" ``` ## Error Handling Handle API errors gracefully: ```python from datadog_api_client import ApiClient, Configuration from datadog_api_client.v2.api.monitors_api import MonitorsApi from datadog_api_client.exceptions import ApiException configuration = Configuration() try: with ApiClient(configuration) as api_client: monitors_api = MonitorsApi(api_client) monitors = monitors_api.list_monitors() print(f"Found {len(monitors.data)} monitors") except ApiException as e: print(f"API Error: {e.status} - {e.reason}") print(f"Details: {e.body}") except Exception as e: print(f"Unexpected error: {e}") ``` ## Using with Virtual Environments Recommended for Python projects: ```bash # Create virtual environment python3 -m venv datadog_env # Activate source datadog_env/bin/activate # macOS/Linux datadog_env\Scripts\activate # Windows # Install package pip install datadog-api-client # Set environment variables in .env file echo "DD_API_KEY=your_key" > .env echo "DD_APP_KEY=your_app_key" >> .env # Load before running scripts export $(cat .env | xargs) python your_script.py ``` ## Async Operations For asynchronous code (requires `pip install datadog-api-client[async]`): ```python from datadog_api_client import AsyncApiClient, Configuration from datadog_api_client.v2.api.monitors_api import MonitorsApi import asyncio async def get_monitors(): configuration = Configuration() async with AsyncApiClient(configuration) as api_client: monitors_api = MonitorsApi(api_client) monitors = await monitors_api.list_monitors() return monitors # Run the async function monitors = asyncio.run(get_monitors()) print(monitors) ``` ## Troubleshooting ### Issue: "Unauthorized" (401 Error) ``` Datadog API error: 401 UNAUTHORIZED ``` **Solutions:** - Verify `DD_API_KEY` is set correctly: `echo $DD_API_KEY` - Verify `DD_APP_KEY` is set correctly: `echo $DD_APP_KEY` - Keys haven't rotated or been revoked - API key has sufficient permissions ### Issue: "Not Found" (404 Error) ``` Datadog API error: 404 NOT_FOUND ``` **Solutions:** - Check the correct site/region: `datadoghq.com` vs `datadoghq.eu` - Verify the resource ID exists - Ensure you're using the correct API endpoint ### Issue: "Too Many Requests" (429 Error) ``` Datadog API error: 429 TOO_MANY_REQUESTS ``` **Solutions:** ```python configuration.max_retries = 5 # Enable automatic retries ``` ### Issue: Import Errors ``` ModuleNotFoundError: No module named 'datadog_api_client' ``` **Solutions:** - Install the package: `pip install datadog-api-client` - Verify correct virtual environment: `which python` - Check installation: `pip show datadog-api-client` ## Next Steps 1. **Read the full API reference:** See `API_REFERENCE.md` 2. **Explore examples:** Check the GitHub repository for v1/v2 examples 3. **Review Datadog docs:** https://docs.datadoghq.com/ 4. **Join the community:** Slack, GitHub Discussions ## Resources - **GitHub:** https://github.com/DataDog/datadog-api-client-python - **PyPI:** https://pypi.org/project/datadog-api-client/ - **Official API Docs:** https://docs.datadoghq.com/api/latest/?tab=python - **API Client Docs:** https://datadoghq.dev/datadog-api-client-python/ --- # Datadog Python API Client - Models and Types Reference **Comprehensive guide to common model classes and data types** ## Table of Contents 1. [Monitor Models](#monitor-models) 2. [Log Models](#log-models) 3. [APM Models](#apm-models) 4. [Dashboard Models](#dashboard-models) 5. [Common Patterns](#common-patterns) 6. [Error Models](#error-models) --- ## Monitor Models ### Monitor Type Enum representing monitor types: ```python from datadog_api_client.v2.model.monitor_type import MonitorType MonitorType.COMPOSITE_MONITOR # Composite monitor MonitorType.LOG_ALERT # Log alert MonitorType.METRIC_ALERT # Metric alert MonitorType.SERVICE_CHECK # Service check MonitorType.SYNTHETICS_ALERT # Synthetics alert MonitorType.EVENT_ALERT # Event alert MonitorType.PROCESS_ALERT # Process alert MonitorType.SLO_ALERT # SLO alert ``` ### Monitor Class The main monitor object: ```python from datadog_api_client.v2.model.monitor import Monitor monitor = Monitor( name="CPU Alert", # Display name type=MonitorType.METRIC_ALERT, # Monitor type query="avg(last_5m):avg:system.cpu{*} > 0.8", # Alert condition message="CPU is high: {{value}}%", # Notification message tags=["production", "critical"], # Tags enabled=True, # Enable/disable priority=1, # Priority (1-5) no_data_timeframe=10, # Minutes without data before alert evaluation_delay=300, # Evaluation delay in seconds keep_alive_interval=600, # Alert persistence interval ) ``` ### Monitor Attributes Key properties: ```python # After creating/retrieving a monitor monitor.id # Integer, unique identifier monitor.name # String, display name monitor.type # MonitorType enum monitor.query # String, alert condition monitor.message # String, notification message monitor.tags # List[String], associated tags monitor.enabled # Boolean, enable status monitor.created_at # Integer, Unix timestamp monitor.modified_at # Integer, Unix timestamp monitor.created_by # Monitor creator info monitor.modified_by # Last modifier info monitor.state # MonitorState (alert, no_alert, ok) ``` ### Monitor State The state of a monitor: ```python from datadog_api_client.v2.model.monitor_state import MonitorState MonitorState.ALERT # Monitor is in alert state MonitorState.NO_ALERT # Monitor is not alerting MonitorState.UNKNOWN # State unknown ``` --- ## Log Models ### LogsListRequest Request object for querying logs: ```python from datadog_api_client.v2.model.logs_list_request import LogsListRequest request = LogsListRequest( filter={ "from": "2024-01-01T00:00:00Z", # Start date (ISO 8601) "to": "2024-01-31T23:59:59Z", # End date (ISO 8601) "query": "status:error source:api", # Query string }, page={ "limit": 100, # Results per page (max 1000) "cursor": None, # Pagination cursor }, sort=Sort.TIMESTAMP_DESCENDING, # Sort order ) ``` ### Log Attributes Properties of a log entry: ```python log.id # Unique log ID log.attributes.host # Hostname log.attributes.service # Service name log.attributes.status # Log level (debug, info, warn, error) log.attributes.message # Log message log.attributes.tags # Associated tags log.attributes.timestamp # Log timestamp (Unix milliseconds) log.attributes.source # Log source (application, language) ``` ### Log Aggregation Group logs by attributes: ```python from datadog_api_client.v2.model.logs_aggregate_request import LogsAggregateRequest request = LogsAggregateRequest( filter={ "from": "2024-01-01T00:00:00Z", "to": "2024-01-31T23:59:59Z", "query": "status:error", }, compute=[ LogsMetricCompute( aggregation="count", # count, cardinality, percentile, etc. metric="", # Leave empty for count type=AggregationMetricType.COUNT, ) ], group_by=[ LogsGroupBy(path="@status", limit=10) # Group by status ], ) ``` --- ## APM Models ### Service Represents an APM service: ```python service.name # Service name service.type # Service type (web, db, cache, etc.) service.version # Service version service.environment # Environment (prod, staging, dev) service.stats.throughput # Requests per second service.stats.latency # Response time in milliseconds service.stats.errors # Error rate percentage service.stats.apdex # Apdex score (0-1) ``` ### Span Represents a distributed trace span: ```python span.id # Unique span ID span.trace_id # Associated trace ID span.parent_id # Parent span ID span.service # Service name span.operation_name # Operation being traced span.resource_name # Resource identifier span.start_time # Start timestamp (nanoseconds) span.duration # Duration in nanoseconds span.tags # Associated tags and metadata span.metrics # Span metrics ``` --- ## Dashboard Models ### Dashboard Main dashboard object: ```python from datadog_api_client.v2.model.dashboard import Dashboard from datadog_api_client.v2.model.dashboard_type import DashboardType dashboard = Dashboard( title="System Overview", # Dashboard title description="Key system metrics", # Description type=DashboardType.TIMEBOARD, # Dashboard type tags=["system", "monitoring"], # Tags is_read_only=False, # Read-only flag layout_type=LayoutType.ORDERED, # Layout type widgets=[], # Dashboard widgets ) ``` ### Dashboard Widget Individual dashboard widget: ```python from datadog_api_client.v2.model.widget import Widget from datadog_api_client.v2.model.widget_definition import WidgetDefinition widget = Widget( definition=WidgetDefinition( type=WidgetType.TIMESERIES, # Widget type requests=[ WidgetRequest( response_format="timeseries", on_right_yaxis=False, queries=[ MetricsQueryOptions( query="avg:system.cpu{*}", # Metric query ) ], ) ], title="CPU Usage", # Widget title yaxis=YAxis( min=0, # Y-axis minimum max=100, # Y-axis maximum ), ) ) ``` ### Widget Types Common widget types: ```python from datadog_api_client.v2.model.widget_type import WidgetType WidgetType.TIMESERIES # Time series graph WidgetType.QUERY_VALUE # Single value display WidgetType.GAUGE # Gauge visualization WidgetType.DISTRIBUTION # Distribution chart WidgetType.HEATMAP # Heatmap visualization WidgetType.TOPLIST # Top list WidgetType.TABLE # Data table WidgetType.NUMBER # Number display WidgetType.ALERT_GRAPH # Alert status graph ``` --- ## Common Patterns ### Pagination with Response ```python from datadog_api_client.v2.api.monitors_api import MonitorsApi with ApiClient(configuration) as api_client: api = MonitorsApi(api_client) # First page response = api.list_monitors(page_size=10) # Access data for monitor in response.data: print(monitor.name) # Check for more pages if hasattr(response, 'links') and 'next' in response.links: # Fetch next page using cursor/offset next_response = api.list_monitors( page_size=10, page_number=2 ) ``` ### Filtering and Sorting ```python # Monitors with filter monitors = api.list_monitors( name="CPU", # Filter by name tags=["production"], # Filter by tags status="alert", # Filter by status ) # Sort ascending/descending monitors = api.list_monitors( sort="name", # Sort by field # Use "-name" for descending ) ``` ### Optional Parameters ```python # Include optional fields response = api.list_dashboards( filter_shared=False, # Exclude shared dashboards sort_field="name", # Sort by field sort_direction="asc", # Sort direction ) ``` ### Bulk Operations ```python # Create multiple monitors monitors = [] for i in range(3): monitor = Monitor( name=f"Monitor {i}", type=MonitorType.METRIC_ALERT, query=f"avg(last_5m):avg:system.load{{host:{i}}} > 0.5", message="Load is high", ) response = api.create_monitor(body=monitor) monitors.append(response) ``` ### Update Operations ```python # Update an existing monitor monitor.message = "Updated message" monitor.tags = ["updated-tag"] updated = api.update_monitor( monitor_id=monitor.id, body=monitor ) ``` ### Delete Operations ```python # Delete a monitor api.delete_monitor(monitor_id=123) # Delete with force flag (for composite monitors) api.delete_monitor( monitor_id=123, force=True ) ``` --- ## Error Models ### API Exception Handle API errors: ```python from datadog_api_client.exceptions import ApiException try: response = api.create_monitor(body=monitor) except ApiException as e: print(f"Status: {e.status}") # HTTP status code print(f"Reason: {e.reason}") # HTTP reason print(f"Body: {e.body}") # Response body print(f"Headers: {e.headers}") # Response headers ``` ### Common HTTP Status Codes ```python 400 # Bad Request - Invalid parameters 401 # Unauthorized - Invalid credentials 403 # Forbidden - Insufficient permissions 404 # Not Found - Resource doesn't exist 409 # Conflict - Resource already exists 429 # Too Many Requests - Rate limit exceeded 500 # Internal Server Error 503 # Service Unavailable ``` ### Error Response Body Typical error response: ```python { "errors": [ "Invalid parameter: query must not be empty" ] } ``` ### Validation Errors Validate input before sending: ```python # Ensure required fields if not monitor.name or not monitor.query: raise ValueError("Monitor name and query are required") # Validate types if not isinstance(monitor.type, MonitorType): raise ValueError("Invalid monitor type") ``` --- ## Type Hints and IDE Support The library uses Python type hints for better IDE support: ```python from datadog_api_client.v2.api.monitors_api import MonitorsApi from datadog_api_client.v2.model.monitor import Monitor from datadog_api_client.v2.model.monitors_list_response import MonitorsListResponse def create_cpu_monitor(api: MonitorsApi, threshold: float) -> Monitor: """Create a CPU usage monitor""" monitor = Monitor( name="CPU Alert", type=MonitorType.METRIC_ALERT, query=f"avg(last_5m):avg:system.cpu{{*}} > {threshold}", message="CPU is high", ) return api.create_monitor(body=monitor) def list_all_monitors(api: MonitorsApi) -> list[Monitor]: """Get all monitors""" response: MonitorsListResponse = api.list_monitors() return response.data ``` --- ## Working with Enums Safe enum usage: ```python from datadog_api_client.v2.model.monitor_type import MonitorType from datadog_api_client.v2.model.monitor_state import MonitorState # Direct enum comparison if monitor.state == MonitorState.ALERT: print("Monitor is alerting") # Get enum value string monitor_type_str = monitor.type.value # "metric_alert" # Create from string monitor_type = MonitorType(user_input) ``` --- ## Serialization Convert models to/from JSON: ```python import json # Model to dictionary monitor_dict = monitor.to_dict() # Model to JSON string monitor_json = json.dumps(monitor_dict, default=str) # Dictionary to model from datadog_api_client.v2.model.monitor import Monitor monitor = Monitor.from_dict(monitor_dict) ``` --- ## References - **API Spec:** https://docs.datadoghq.com/api/latest/?tab=python - **GitHub:** https://github.com/DataDog/datadog-api-client-python - **Model Documentation:** https://datadoghq.dev/datadog-api-client-python/