dt-async - Dynatrace Python API Client

dt-async is a Python client for the Dynatrace Rest API.
It focuses on ease of use and nice type hints, perfect to explore the API and create quick scripts

This project is a modified fork of the original library released by Dynatrace.

Install

$ pip install dt-async

Authentication

This library supports two authentication methods:

OAuth 2.0 (client credentials)

Use DynatraceOAuthCredentials for the OAuth 2.0 client credentials flow.

from dynatrace import DynatraceAsync, DynatraceOAuthCredentials

dt = DynatraceAsync(
    base_url="environment_url",
    credentials=DynatraceOAuthCredentials(
        client_id="oauth_client_id",
        client_secret="oauth_client_secret",
        account_uuid="your-account-uuid",
        scope="environment-api:entities:read environment-api:metrics:read",
    ),
)

The scope value must include the permissions required by the APIs you want to call.

API Token

Use DynatraceAccessToken for token-based authentication. The client will send an Api-Token header with every request.

from dynatrace import DynatraceAsync, DynatraceAccessToken

dt = DynatraceAsync(
    base_url="environment_url",
    credentials=DynatraceAccessToken(token="dt0c01..."),
)

Simple Demo

import asyncio
from datetime import datetime, timedelta

from dynatrace import DynatraceAsync, DynatraceOAuthCredentials
from dynatrace import TOO_MANY_REQUESTS_WAIT
from dynatrace.environment_v2.credential_vault import PublicCertificateCredentials
from dynatrace.environment_v2.settings import SettingsObjectCreate
from dynatrace.environment_v2.tokens_api import (
    SCOPE_METRICS_INGEST,
    SCOPE_METRICS_READ,
)

async def main():
    # Create a Dynatrace client with OAuth credentials
    async with DynatraceAsync(
        base_url="environment_url",
        credentials=DynatraceOAuthCredentials(
            client_id="oauth_client_id",
            client_secret="oauth_client_secret",
            account_uuid="your-account-uuid",
        ),
    ) as dt:
        # Create a client that handles too many requests (429)
        # dt = DynatraceAsync(
        #     base_url="environment_url",
        #     credentials=DynatraceOAuthCredentials(
        #         client_id="oauth_client_id",
        #         client_secret="oauth_client_secret",
        #         account_uuid="your-account-uuid",
        #     ),
        #     too_many_requests_strategy=TOO_MANY_REQUESTS_WAIT,
        # )
        # Create a client that automatically retries on errors, up to 5 times, with a 1 second delay between retries
        # dt = DynatraceAsync(
        #     base_url="environment_url",
        #     credentials=DynatraceOAuthCredentials(
        #         client_id="oauth_client_id",
        #         client_secret="oauth_client_secret",
        #         account_uuid="your-account-uuid",
        #     ),
        #     retries=5,
        #     retry_delay_ms=1000,
        # )

        # Create a client with a custom HTTP timeout of 10 seconds
        # dt = DynatraceAsync(
        #     base_url="environment_url",
        #     credentials=DynatraceOAuthCredentials(
        #         client_id="oauth_client_id",
        #         client_secret="oauth_client_secret",
        #         account_uuid="your-account-uuid",
        #     ),
        #     timeout=10,
        # )

        # Create a client that automatically retries on errors, up to 5 times, with a 1 second delay between retries
        # dt = DynatraceAsync(
        #     client_id="oauth_client_id",
        #     client_secret="oauth_client_secret",
        #     account_uuid="your-account-uuid",
        #     base_url="environment_url",
        #     retries=5,
        #     retry_delay_ms=1000,
        # )

        # Create a client with a custom HTTP timeout of 10 seconds
        # dt = DynatraceAsync(
        #     client_id="oauth_client_id",
        #     client_secret="oauth_client_secret",
        #     account_uuid="your-account-uuid",
        #     base_url="environment_url",
        #     timeout=10,
        # )

        # Get all hosts and some properties
        async for entity in await dt.entities.list(
            'type("HOST")',
            fields="properties.memoryTotal,properties.monitoringMode",
        ):
            print(entity.entity_id, entity.display_name, entity.properties)

        # Get idle CPU for all hosts
        async for metric in await dt.metrics.query(
            "builtin:host.cpu.idle",
            resolution="Inf",
        ):
            print(metric)

        # Print dimensions, timestamp and values for the AWS Billing Metric
        async for metric in await dt.metrics.query(
            "ext:cloud.aws.billing.estimatedChargesByRegionCurrency"
        ):
            for data in metric.data:
                for timestamp, value in zip(data.timestamps, data.values):
                    print(data.dimensions, timestamp, value)

        # Get all ActiveGates
        async for ag in await dt.activegates.list():
            print(ag)

        # Get metric descriptions for all host metrics
        async for metric in await dt.metrics.list("builtin:host.*"):
            print(metric)

        # Delete endpoints that contain the word test
        async for plugin in await dt.plugins.list():
            # This could also be dt.get_endpoints(plugin.id)
            async for endpoint in plugin.endpoints:
                if "test" in endpoint.name:
                    await endpoint.delete(plugin.id)

        # Prints dashboard ID, owner and number of tiles
        async for dashboard in await dt.dashboards.list():
            full_dashboard = await dashboard.get_full_dashboard()
            print(full_dashboard.id, dashboard.owner, len(full_dashboard.tiles))

        # Delete API Tokens that haven't been used for more than 3 months
        async for token in await dt.tokens.list(fields="+lastUsedDate,+scopes"):
            if token.last_used_date and token.last_used_date < datetime.now() - timedelta(
                days=90
            ):
                print(
                    f"Deleting token! {token}, last used date: {token.last_used_date}"
                )

        # Create an API Token that can read and ingest metrics
        new_token = await dt.tokens.create(
            "metrics_token",
            scopes=[SCOPE_METRICS_READ, SCOPE_METRICS_INGEST],
        )
        print(new_token.token)

        # Upload a public PEM certificate to the Credential Vault
        with open("ca.pem", "r") as f:
            ca_cert = f.read()

        my_cred = PublicCertificateCredentials(
            name="my_cred",
            scopes=["EXTENSION"],
            description="my_cred description",
            owner_access_only=False,
            certificate=ca_cert,
            password="",
            certificate_format="PEM",
        )

        credential = await dt.credentials.post(my_cred)
        print(credential.id)

        # Create a new settings 2.0 object
        settings_value = {
            "enabled": True,
            "summary": "DT API TEST 1",
            "queryDefinition": {
                "type": "METRIC_KEY",
                "metricKey": "netapp.ontap.node.fru.state",
                "aggregation": "AVG",
                "entityFilter": {
                    "dimensionKey": "dt.entity.netapp_ontap:fru",
                    "conditions": [],
                },
                "dimensionFilter": [],
            },
            "modelProperties": {
                "type": "STATIC_THRESHOLD",
                "threshold": 100.0,
                "alertOnNoData": False,
                "alertCondition": "BELOW",
                "violatingSamples": 3,
                "samples": 5,
                "dealertingSamples": 5,
            },
            "eventTemplate": {
                "title": "OnTap {dims:type} {dims:fru_id} is in Error State",
                "description": "OnTap field replaceable unit (FRU) {dims:type} with id {dims:fru_id} on node {dims:node} in cluster {dims:cluster} is in an error state.\n",
                "eventType": "RESOURCE",
                "davisMerge": True,
                "metadata": [],
            },
            "eventEntityDimensionKey": "dt.entity.netapp_ontap:fru",
        }

        settings_object = SettingsObjectCreate(
            schema_id="builtin:anomaly-detection.metric-events",
            value=settings_value,
            scope="environment",
        )
        await dt.settings.create_object(validate_only=False, body=settings_object)


asyncio.run(main())

Implementation Progress

Environment API V2

API	Level	Access
Access Tokens - API tokens	✔️	dt.tokens
Access tokens - Tenant tokens	✔️	dt.tenant_tokens
ActiveGates	✔️	dt.activegates
ActiveGates - Auto-update configuration	✔️	dt.activegates_autoupdate_configuration
ActiveGates - Auto-update jobs	✔️	dt.activegates_autoupdate_jobs
ActiveGates - Remote configuration	✔️	dt.activegates_remote_configuration
Audit Logs	✔️	dt.audit_logs
Events	⚠️	dt.events_v2
Extensions 2.0	✔️	dt.extensions_v2
Logs	⚠️	dt.logs
Metrics	✔️	dt.metrics
Monitored entities	⚠️	dt.entities
Monitored entities - Custom tags	✔️	dt.custom_tags
Network zones	⚠️	dt.network_zones
OneAgents - Remote configuration	✔️	dt.oneagents_remote_configuration
Problems	✔️	dt.problems
Security problems	✔️	dt.security_problems
Service-level objectives	✔️	dt.slos
Settings	⚠️	dt.settings
Credential vault	✔️	dt.credentials

Environment API V1

API	Level	Access
Anonymization	❌
Cluster time	✔️	dt.cluster_time
Cluster version	❌
Custom devices	✔️	dt.custom_devices
Deployment	✔️	dt.deployment
Events	⚠️	dt.events
JavaScript tag management	❌
Log monitoring - Custom devices	❌
Log monitoring - Hosts	❌
Log monitoring - Process groups	❌
Maintenance window	❌
OneAgent on a host	⚠️	dt.oneagents
Problem	❌
Synthetic - Locations and nodes	❌
Synthetic - Monitors	⚠️	dt.synthetic_monitors
Synthetic - Third party	✔️	dt.third_part_synthetic_tests
Threshold	❌
Timeseries	⚠️	dt.timeseries
Tokens	❌
Topology & Smartscape - Application	❌
Topology & Smartscape - Custom device	⚠️	dt.custom_devices
Topology & Smartscape - Host	⚠️	dt.smartscape_hosts
Topology & Smartscape - Process	❌
Topology & Smartscape - Process group	❌
Topology & Smartscape - Service	❌
User sessions	❌

Configuration API V1

API	Level	Access
Alerting Profiles	⚠️	dt.alerting_profiles
Anomaly detection - Applications	❌
Anomaly detection - AWS	❌
Anomaly detection - Database services	❌
Anomaly detection - Disk events	❌
Anomaly detection - Hosts	❌
Anomaly detection - Metric events	⚠️	dt.anomaly_detection_metric_events
Anomaly detection - Process groups	⚠️	dt.anomaly_detection_process_groups
Anomaly detection - Services	❌
Anomaly detection - VMware	❌
Automatically applied tags	⚠️	dt.auto_tags
AWS credentials configuration	❌
AWS PrivateLink	❌
Azure credentials configuration	❌
Calculated metrics - Log monitoring	❌
Calculated metrics - Mobile & custom applications	❌
Calculated metrics - Services	❌
Calculated metrics - Synthetic	❌
Calculated metrics - Web applications	❌
Cloud Foundry credentials configuration	❌
Conditional naming	❌
Custom tags	✔️	dt.custom_tags
Dashboards	⚠️	dt.dashboards
Data privacy and security	❌
Extensions	✔️	dt.extensions
Frequent issue detection	❌
Kubernetes credentials configuration	❌
Maintenance windows	⚠️	dt.maintenance_windows
Management zones	⚠️	dt.management_zones
Notifications	⚠️	dt.notifications
OneAgent - Environment-wide configuration	✔️	dt.oneagents_config_environment
OneAgent in a host group	✔️	dt.oneagents_config_hostgroup
OneAgent on a host	✔️	dt.oneagents_config_host
Plugins	⚠️	dt.plugins
Remote environments	❌
Reports	❌
RUM - Allowed beacon origins for CORS	❌
RUM - Application detection rules	❌
RUM - Application detection rules - Host detection	❌
RUM - Content resources	❌
RUM - Geographic regions - custom client IP headers	❌
RUM - Geographic regions - IP address mapping	❌
RUM - Mobile and custom application configuration	❌
RUM - Web application configuration	❌
Service - Custom services	❌
Service - Detection full web request	❌
Service - Detection full web service	❌
Service - Detection opaque and external web request	❌
Service - Detection opaque and external web service	❌
Service - Failure detection parameter sets	❌
Service - Failure detection rules	❌
Service - IBM MQ tracing	❌
Service - Request attributes	❌
Service - Request naming	❌

Platform API

API	Level	Access
AppEngine - Registry - Apps	✔️	dt.platform.appengine_registry
AppEngine - Registry - Schema Manifest	✔️	dt.platform.appengine_registry
Davis CoPilot API	✔️	dt.platform.davis_copilot
Davis AI - Predictive and Causal	⚠️	dt.platform.davis_analyzers
DQL Query	✔️	dt.platform.grail_query_execution, dt.platform.grail_query_assistance

Account API

API	Level	Access
Account Management - Account limits	❌
Account Management - User management	✔️	dt.account.iam_users
Account Management - Group management	✔️	dt.account.iam_groups
Account Management - Permission management	✔️	dt.account.iam_policies
Account Management - Policy management	✔️	dt.account.iam_policies
Account Management - Service user management	✔️	dt.account.iam_service_users
Account Management - Platform tokens	✔️	dt.account.iam_platform_tokens
Environment management API v1	✔️	dt.account.env_v1
Environment management API v2	✔️	dt.account.env_v2
Dynatrace Platform Subscription - Subscription management	✔️	dt.account.sub_v2
Dynatrace Platform Subscription - Subscription environments v2	❌
Dynatrace Platform Subscription - Subscription environments v3	✔️	dt.account.sub_v3
Dynatrace Platform Subscription - Rate cards	✔️	dt.account.sub_v1_rate_cards
Dynatrace Platform Subscription - Cost allocation	✔️	dt.account.sub_v1_cost_allocation
Account Settings	❌
Account Audits	✔️	dt.account.audits
Reference data	❌
Notifications	✔️	dt.account.notifications