"""Datadog Connector.""" from contextlib import contextmanager import config from datadog import api, initialize @contextmanager def datadog_connection(): """Context manager for datadog connection objects. Yields: datadog.api: connection to datadog api """ dd_api_key = getattr(config, 'DD_API_KEY', None) dd_app_key = getattr(config, 'DD_APP_KEY', None) if not all(isinstance(k, str) and k for k in (dd_api_key, dd_app_key)): raise ValueError('Both DD_API_KEY and DD_APP_KEY must be non-empty strings.') initialize(api_key=dd_api_key, app_key=dd_app_key) yield api def publish_metric(metric: str, value: int = 1) -> None: """Publish a metric to DataDog.""" if config.ENVIRONMENT.lower() in [config.DEV_ENVIRONMENT, config.TEST_ENVIRONMENT]: return with datadog_connection() as datadog: datadog.Metric.send(metric=metric, points=value, tags=[ f'environment:{config.ENVIRONMENT}', f'service_name:{config.SERVICE_NAME}', ])