API Integration

Connecting to REST APIs

Most Python extensions exist to pull data from REST APIs. This module covers authentication patterns, error handling, and metric reporting from API responses.

Basic API Collection Pattern

@Extension.schedule(period=60)
def collect_data(self):
    url = self.config.get("url")
    token = self.config.get("api_token")

    try:
        response = self.http_client.get(
            f"{url}/api/v1/system/status",
            headers={"Authorization": f"Bearer {token}"},
            timeout=30,
        )
        response.raise_for_status()
        data = response.json()
    except Exception as e:
        self.logger.error(f"API call failed: {e}")
        return

    self.report_metric(
        key="com.dynatrace.extension.myapi.cpu",
        value=data.get("cpu_percent", 0),
        dimensions={"device.address": url, "device.name": data.get("hostname", "unknown")},
    )

Authentication Patterns

API Token (Header)

headers = {"Authorization": f"Bearer {self.config.get('api_token')}"}
response = self.http_client.get(url, headers=headers)

Basic Auth

from requests.auth import HTTPBasicAuth
auth = HTTPBasicAuth(self.config.get("username"), self.config.get("password"))
response = self.http_client.get(url, auth=auth)

Session-Based (Login First)

def initialize(self):
    self.session_token = None

def _login(self):
    """Get session token from login endpoint."""
    resp = self.http_client.post(
        f"{self.config.get('url')}/api/login",
        json={"user": self.config.get("username"), "pass": self.config.get("password")},
    )
    self.session_token = resp.json().get("token")

@Extension.schedule(period=60)
def collect_data(self):
    if not self.session_token:
        self._login()
    headers = {"X-Session-Token": self.session_token}
    resp = self.http_client.get(f"{self.config.get('url')}/api/data", headers=headers)
    if resp.status_code == 401:
        self._login()  # Token expired, re-login
        headers = {"X-Session-Token": self.session_token}
        resp = self.http_client.get(f"{self.config.get('url')}/api/data", headers=headers)

Multiple Endpoints Pattern

The monitoring configuration can define multiple endpoints (devices). Loop through them:

@Extension.schedule(period=60)
def collect_data(self):
    for endpoint in self.config.get("endpoints", []):
        url = endpoint.get("url")
        name = endpoint.get("name", url)
        try:
            self._collect_device(url, name)
        except Exception as e:
            self.logger.error(f"Failed to collect from {name}: {e}")

def _collect_device(self, url, name):
    resp = self.http_client.get(f"{url}/api/status")
    data = resp.json()
    dims = {"device.address": url, "device.name": name}

    self.report_metric(key="myext.cpu", value=data["cpu"], dimensions=dims)
    self.report_metric(key="myext.memory", value=data["mem"], dimensions=dims)

Handling Paginated APIs

def _get_all_items(self, url, headers):
    items = []
    page = 1
    while True:
        resp = self.http_client.get(f"{url}?page={page}&limit=100", headers=headers)
        data = resp.json()
        items.extend(data.get("items", []))
        if not data.get("has_more", False):
            break
        page += 1
    return items

Error Handling Best Practices

• Always wrap API calls in try/except — a crash kills the extension process
• Log errors with context — include URL, status code, response body snippet
• Use timeouts — default is no timeout, which can hang the extension
• Handle partial failures — if one endpoint fails, continue to the next
• Don't store secrets in code — use monitoring configuration fields

Real Bug: Missing __main__.py

In our Waze Police v0.0.1, the extension failed to start because the wheel package was missing __main__.py. The EEC log showed:

ERROR: No module named 'waze_police.__main__'

The fix: ensure your package has __main__.py with the main() entry point. The EEC runs python -m your_package, which requires this file.