Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -341,6 +341,6 @@ asyncio.run(main())
Dynatrace Platform Subscription - Rate cards | :heavy_check_mark: | `dt.account.sub_v1_rate_cards` |
Dynatrace Platform Subscription - Cost allocation | :heavy_check_mark: | `dt.account.sub_v1_cost_allocation` |
Account Settings | :x: | |
Account Audits | :x: | |
Account Audits | :heavy_check_mark: | `dt.account.audits` |
Reference data | :x: | |
Notifications | :x: | |
Notifications | :heavy_check_mark: | `dt.account.notifications` |
92 changes: 92 additions & 0 deletions dynatrace/account/audits.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,92 @@
"""Account audits API wrappers."""

import builtins
from datetime import datetime
from typing import Any

from dynatrace.dynatrace_object import DynatraceObject
from dynatrace.http_client import HttpClient
from dynatrace.utils import timestamp_to_string


class AccountAuditsService:
"""/audit/v1 Account audits API."""

def __init__(self, http_client: HttpClient) -> None:
self.__http_client = http_client

async def list(
self,
account_uuid: str,
start_time: datetime | str | None = None,
end_time: datetime | str | None = None,
add_fields: builtins.list[str] | None = None,
filter: str | None = None,
limit: int | None = None,
scan_limit_gigabyte: int | None = None,
result_size_limit_megabyte: int | None = None,
) -> "AuditsByAccount":
params: dict[str, Any] = {
"startTime": timestamp_to_string(start_time),
"endTime": timestamp_to_string(end_time),
"filter": filter,
"limit": limit,
"scanLimitGigabyte": scan_limit_gigabyte,
"resultSizeLimitMegabyte": result_size_limit_megabyte,
}
if add_fields:
params["addFields"] = ",".join(add_fields)

resp = (
await self.__http_client.make_request(
f"/audit/v1/accounts/{account_uuid}",
params=params,
)
).json()
return AuditsByAccount(raw_element=resp)


class Audit(DynatraceObject):
def _create_from_raw_data(self, raw_element: dict[str, Any]):
self.event_id: str | None = raw_element.get("eventId")
self.timestamp: str | None = raw_element.get("timestamp")
self.user: str | None = raw_element.get("user")
self.resource: str | None = raw_element.get("resource")
self.resource_name: str | None = raw_element.get("resourceName")
self.event_provider: str | None = raw_element.get("eventProvider")
self.event_type: str | None = raw_element.get("eventType")
self.account_uuid: str | None = raw_element.get("accountUuid")
self.authentication_client_id: str | None = raw_element.get(
"authenticationClientId"
)
self.authentication_grant_type: str | None = raw_element.get(
"authenticationGrantType"
)
self.authentication_token: str | None = raw_element.get("authenticationToken")
self.authentication_type: str | None = raw_element.get("authenticationType")
self.details: dict[str, str] | None = raw_element.get("details")
self.event_outcome: str | None = raw_element.get("eventOutcome")
self.event_reason: str | None = raw_element.get("eventReason")
self.event_version: str | None = raw_element.get("eventVersion")
self.origin_address: str | None = raw_element.get("originAddress")
self.origin_session: str | None = raw_element.get("originSession")
self.origin_type: str | None = raw_element.get("originType")
self.origin_x_forwarded_for: str | None = raw_element.get("originXForwardedFor")
self.resource_id: str | None = raw_element.get("resourceId")
self.environment_uuid: str | None = raw_element.get("environmentUuid")
self.user_organization: str | None = raw_element.get("userOrganization")


class AuditWarning(DynatraceObject):
def _create_from_raw_data(self, raw_element: dict[str, Any]):
self.message: str | None = raw_element.get("message")


class AuditsByAccount(DynatraceObject):
def _create_from_raw_data(self, raw_element: dict[str, Any]):
self.audits: builtins.list[Audit] = [
Audit(raw_element=a) for a in raw_element.get("audits", [])
]
self.warnings: builtins.list[AuditWarning] = [
AuditWarning(raw_element=w) for w in raw_element.get("warnings", [])
]
114 changes: 114 additions & 0 deletions dynatrace/account/notifications.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,114 @@
"""Account notifications API wrappers."""

from __future__ import annotations

import builtins
from datetime import datetime
from enum import Enum
from typing import Any

from dynatrace.dynatrace_object import DynatraceObject
from dynatrace.http_client import HttpClient
from dynatrace.utils import timestamp_to_string


class NotificationService:
"""/v1/accounts/{accountUuid}/notifications API."""

def __init__(self, http_client: HttpClient) -> None:
self.__http_client = http_client

async def list(
self,
account_uuid: str,
start_date_time: datetime | str | None = None,
end_date_time: datetime | str | None = None,
types: builtins.list[str | NotificationType] | None = None,
severities: builtins.list[str | NotificationSeverity] | None = None,
capabilities: builtins.list[str] | None = None,
environments: builtins.list[str] | None = None,
page: int | None = None,
page_size: int | None = None,
sorts: builtins.list[str | NotificationSortField] | None = None,
) -> NotificationList:
body: dict[str, Any] = {
"startDateTime": timestamp_to_string(start_date_time),
"endDateTime": timestamp_to_string(end_date_time),
"types": [self._enum_value(item) for item in types] if types else None,
"severities": (
[self._enum_value(item) for item in severities] if severities else None
),
"capabilities": capabilities,
"environments": environments,
"page": page,
"pageSize": page_size,
"sorts": [self._enum_value(item) for item in sorts] if sorts else None,
}

resp = (
await self.__http_client.make_request(
f"/v1/accounts/{account_uuid}/notifications",
method="POST",
params=body,
)
).json()
return NotificationList(raw_element=resp)

@staticmethod
def _enum_value(value: str | Enum) -> str:
return value.value if isinstance(value, Enum) else value


class NotificationType(Enum):
FORECAST = "FORECAST"
BUDGET = "BUDGET"
COST = "COST"
BYOK_REVOKED = "BYOK_REVOKED"
BYOK_ACTIVATED = "BYOK_ACTIVATED"


class NotificationSeverity(Enum):
SEVERE = "SEVERE"
WARN = "WARN"
INFO = "INFO"


class NotificationSortField(Enum):
TYPE = "type"
TYPE_DESC = "-type"
DATE = "date"
DATE_DESC = "-date"


class NotificationDetails(DynatraceObject):
def _create_from_raw_data(self, raw_element: dict[str, Any]):
self.environments: builtins.list[str] | None = raw_element.get("environments")
self.capabilities: builtins.list[str] | None = raw_element.get("capabilities")
self.all_environments: bool | None = raw_element.get("allEnvironments")
self.all_capabilities: bool | None = raw_element.get("allCapabilities")
self.environment_uuid: str | None = raw_element.get("environmentUuid")
self.key_name: str | None = raw_element.get("keyName")


class Notification(DynatraceObject):
def _create_from_raw_data(self, raw_element: dict[str, Any]):
self.key: str | None = raw_element.get("key")
self.account_uuid: str | None = raw_element.get("accountUuid")
self.message: str | None = raw_element.get("message")
self.severity: str | None = raw_element.get("severity")
self.type: str | None = raw_element.get("type")
details = raw_element.get("details") or {}
self.details: NotificationDetails | None = (
NotificationDetails(raw_element=details) if details else None
)
self.date: str | None = raw_element.get("date")


class NotificationList(DynatraceObject):
def _create_from_raw_data(self, raw_element: dict[str, Any]):
self.records: builtins.list[Notification] = [
Notification(raw_element=record)
for record in raw_element.get("records", [])
]
self.total_record_count: int | None = raw_element.get("totalRecordCount")
self.has_next_page: bool | None = raw_element.get("hasNextPage")
8 changes: 8 additions & 0 deletions dynatrace/account/schemas.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
"""High-level account API aggregators."""

from dynatrace.account.audits import AccountAuditsService
from dynatrace.account.env_v1.environments import AccountEnvironmentsV1Service
from dynatrace.account.env_v2.environments import AccountEnvironmentsV2Service
from dynatrace.account.env_v2.settings import SettingService as AccountSettingService
Expand All @@ -8,6 +9,7 @@
from dynatrace.account.iam_v1.policies import AccountPoliciesService
from dynatrace.account.iam_v1.service_users import AccountServiceUsersService
from dynatrace.account.iam_v1.users import AccountUsersService
from dynatrace.account.notifications import NotificationService
from dynatrace.account.sub_v1.cost_allocation import CostAllocationService
from dynatrace.account.sub_v1.rate_cards import RateCardService
from dynatrace.account.sub_v2.subscriptions import SubscriptionService
Expand All @@ -17,6 +19,12 @@

class AccountAPI:
def __init__(self, http_client: HttpClient) -> None:
# Account audit APIs.
self.audits: AccountAuditsService = AccountAuditsService(http_client)

# Account notification APIs.
self.notifications: NotificationService = NotificationService(http_client)

# Account environment management APIs.
self.env_v1: AccountEnvironmentsV1Service = AccountEnvironmentsV1Service(
http_client
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[tool.poetry]
name = "dt-async"
version = "1.0.8"
version = "1.0.9"
description = "Dynatrace async API Python client"
readme = "README.md"
authors = ["David Lopes <davidribeirolopes@gmail.com>"]
Expand Down
108 changes: 108 additions & 0 deletions test/account/test_audits.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,108 @@
import json
from datetime import UTC, datetime
from unittest import mock

from dynatrace import DynatraceAsync
from dynatrace.account.audits import AccountAuditsService, Audit, AuditsByAccount
from dynatrace.http_client import HttpClient


class MockResponse:
def __init__(self, json_data=None):
self._json_data = json_data
self.headers = {}
self.text = json.dumps(json_data) if json_data is not None else ""
self.status_code = 200

def json(self):
return self._json_data


async def test_audits_service_is_exposed(dt: DynatraceAsync):
assert isinstance(dt.account.audits, AccountAuditsService)


async def test_audits_list_returns_models(dt: DynatraceAsync):
account_uuid = "account-123"
start_time = datetime(2026, 1, 1, 0, 0, tzinfo=UTC)
end_time = datetime(2026, 1, 31, 23, 59, tzinfo=UTC)

async def fake_make_request(
self,
path,
params=None,
headers=None,
method="GET",
data=None,
files=None,
query_params=None,
**kwargs,
):
expected_params = {
"startTime": "2026-01-01T00:00:00.000",
"endTime": "2026-01-31T23:59:00.000",
"addFields": "details,eventOutcome",
"filter": "eventType = 'CREATE'",
"limit": 25,
"scanLimitGigabyte": 10,
"resultSizeLimitMegabyte": 5,
}
if (method, path) == ("GET", f"/audit/v1/accounts/{account_uuid}"):
assert params == expected_params
return MockResponse(
{
"audits": [
{
"eventId": "af1f98c9-c611-4056-841b-d039b1af3f98",
"timestamp": "2026-01-01T10:00:00Z",
"user": "user@example.com",
"resource": "POLICY",
"resourceName": "Standard User",
"eventProvider": "Identity & Account Management",
"eventType": "CREATE",
"accountUuid": account_uuid,
"authenticationClientId": "client-1",
"authenticationGrantType": "AUTHORIZATION_CODE",
"authenticationToken": "token-1",
"authenticationType": "OAUTH2",
"details": {
"json_before": '{"enabled": false}',
"json_after": '{"enabled": true}',
},
"eventOutcome": "SUCCESS",
"eventReason": "Created from UI",
"eventVersion": "1.0.0",
"originAddress": "0.0.0.0",
"originSession": "session-1",
"originType": "REST",
"originXForwardedFor": "192.168.1.1",
"resourceId": "resource-1",
"environmentUuid": "environment-1",
"userOrganization": "CUSTOMER",
}
],
"warnings": [{"message": "Your result has been limited to 1."}],
}
)

raise AssertionError(f"Unexpected request: {method} {path}")

with mock.patch.object(HttpClient, "make_request", new=fake_make_request):
audits = await dt.account.audits.list(
account_uuid,
start_time=start_time,
end_time=end_time,
add_fields=["details", "eventOutcome"],
filter="eventType = 'CREATE'",
limit=25,
scan_limit_gigabyte=10,
result_size_limit_megabyte=5,
)

assert isinstance(audits, AuditsByAccount)
assert len(audits.audits) == 1
assert isinstance(audits.audits[0], Audit)
assert audits.audits[0].resource_name == "Standard User"
assert audits.audits[0].details["json_after"] == '{"enabled": true}'
assert len(audits.warnings) == 1
assert audits.warnings[0].message == "Your result has been limited to 1."
Loading
Loading