Include extra useful entry properties when parsing snapshot

dynatrace_extension/__about__.py

@@ -3,4 +3,4 @@
 # SPDX-License-Identifier: MIT
 
 
-__version__ = "1.5.0"
+__version__ = "1.5.1"

dynatrace_extension/cli/create/create.py

@@ -2,7 +2,7 @@
 import re
 import shutil
 from pathlib import Path
-from typing import Final, List, NamedTuple, Tuple
+from typing import Final, NamedTuple
 
 output_mode_folder: Final = 0o755
 output_mode_file: Final = 0o644
@@ -14,7 +14,7 @@ class ReplaceString(NamedTuple):
     replace: str
 
 
-def replace_placeholders(file: Path, replaces: List[Tuple[str, str]]):
+def replace_placeholders(file: Path, replaces: list[tuple[str, str]]):
     with open(file) as f:
         contents = f.read()
     for replace in replaces:

dynatrace_extension/cli/main.py

@@ -4,7 +4,6 @@
 import subprocess
 import sys
 from pathlib import Path
-from typing import List, Optional
 
 import typer
 from dtcli.server_api import upload as dt_cli_upload  # type: ignore
@@ -91,17 +90,17 @@ def build(
         "-k",
         help="Path to the dev fused key-certificate",
     ),
-    target_directory: Optional[Path] = typer.Option(None, "--target-directory", "-t"),
-    extra_platforms: Optional[list[str]] = typer.Option(
+    target_directory: Path | None = typer.Option(None, "--target-directory", "-t"),
+    extra_platforms: list[str] | None = typer.Option(
         None, "--extra-platform", "-e", help="Download wheels for an extra platform"
     ),
-    extra_index_url: Optional[str] = typer.Option(
+    extra_index_url: str | None = typer.Option(
         None, "--extra-index-url", "-i", help="Extra index url to use when downloading dependencies"
     ),
-    find_links: Optional[str] = typer.Option(
+    find_links: str | None = typer.Option(
         None, "--find-links", "-f", help="Extra index url to use when downloading dependencies"
     ),
-    only_extra_platforms: Optional[bool] = typer.Option(
+    only_extra_platforms: bool | None = typer.Option(
         False,
         "--only-extra-platforms",
         "-o",
@@ -163,7 +162,7 @@ def assemble(
 
     # Checks that the module name is valid and exists in the filesystem
     module_folder = Path(extension_dir) / extension_yaml.python.runtime.module
-    src_module_folder = Path('src') / module_folder
+    src_module_folder = Path("src") / module_folder
     if not module_folder.exists() and not src_module_folder.exists():
         msg = f"Extension module folder {module_folder} not found"
         raise FileNotFoundError(msg)
@@ -188,16 +187,16 @@ def assemble(
 @app.command(help="Downloads the dependencies of the extension to the lib folder")
 def wheel(
     extension_dir: Path = typer.Argument(".", help="Path to the python extension"),
-    extra_platforms: Optional[list[str]] = typer.Option(
+    extra_platforms: list[str] | None = typer.Option(
         None, "--extra-platform", "-e", help="Download wheels for an extra platform"
     ),
-    extra_index_url: Optional[str] = typer.Option(
+    extra_index_url: str | None = typer.Option(
         None, "--extra-index-url", "-i", help="Extra index url to use when downloading dependencies"
     ),
-    find_links: Optional[str] = typer.Option(
+    find_links: str | None = typer.Option(
         None, "--find-links", "-f", help="Extra index url to use when downloading dependencies"
     ),
-    only_extra_platforms: Optional[bool] = typer.Option(
+    only_extra_platforms: bool | None = typer.Option(
         False,
         "--only-extra-platforms",
         "-o",
@@ -476,9 +475,7 @@ def ruff_init(extension_dir: Path = typer.Argument(".", help="Path to the python
         console.print(f"Added ruff.toml to {extension_dir.resolve()}", style="bold green")
 
 
-def run_process(
-    command: List[str], cwd: Optional[Path] = None, env: Optional[dict] = None, print_message: Optional[str] = None
-):
+def run_process(command: list[str], cwd: Path | None = None, env: dict | None = None, print_message: str | None = None):
     friendly_command = " ".join(command)
     if print_message is not None:
         console.print(print_message, style="cyan")

dynatrace_extension/sdk/activation.py

@@ -3,7 +3,6 @@
 # SPDX-License-Identifier: MIT
 
 from enum import Enum
-from typing import List
 
 
 class ActivationType(Enum):
@@ -17,7 +16,7 @@ def __init__(self, activation_context_json: dict):
         self.version: str = self._activation_context_json.get("version", "")
         self.enabled: bool = self._activation_context_json.get("enabled", True)
         self.description: str = self._activation_context_json.get("description", "")
-        self.feature_sets: List[str] = self._activation_context_json.get("featureSets", [])
+        self.feature_sets: list[str] = self._activation_context_json.get("featureSets", [])
         self.type: ActivationType = ActivationType.REMOTE if self.remote else ActivationType.LOCAL
         super().__init__()
 

dynatrace_extension/sdk/callback.py

@@ -4,9 +4,9 @@
 
 import logging
 import random
+from collections.abc import Callable
 from datetime import datetime, timedelta
 from timeit import default_timer as timer
-from typing import Callable, Dict, Optional, Tuple
 
 from .activation import ActivationType
 from .communication import MultiStatus, Status, StatusValue
@@ -18,10 +18,10 @@ def __init__(
         interval: timedelta,
         callback: Callable,
         logger: logging.Logger,
-        args: Optional[Tuple] = None,
-        kwargs: Optional[Dict] = None,
+        args: tuple | None = None,
+        kwargs: dict | None = None,
         running_in_sim=False,
-        activation_type: Optional[ActivationType] = None,
+        activation_type: ActivationType | None = None,
     ):
         self.callback: Callable = callback
         if args is None:

dynatrace_extension/sdk/communication.py

@@ -8,16 +8,17 @@
 import logging
 import sys
 from abc import ABC, abstractmethod
+from collections.abc import Generator, Sequence
 from dataclasses import dataclass
 from enum import Enum
 from pathlib import Path
-from typing import Any, Generator, List, Sequence, TypeVar
+from typing import Any, TypeVar
 
 from .vendor.mureq.mureq import HTTPException, Response, request
 
 CONTENT_TYPE_JSON = "application/json;charset=utf-8"
 CONTENT_TYPE_PLAIN = "text/plain;charset=utf-8"
-COUNT_METRIC_ITEMS_DICT = TypeVar("COUNT_METRIC_ITEMS_DICT", str, List[str])
+COUNT_METRIC_ITEMS_DICT = TypeVar("COUNT_METRIC_ITEMS_DICT", str, list[str])
 
 # TODO - I believe these can be adjusted via RuntimeConfig, they can't be constants
 MAX_MINT_LINES_PER_REQUEST = 1000
@@ -60,7 +61,6 @@ def is_error(self) -> bool:
 
 
 class MultiStatus:
-
     def __init__(self):
         self.statuses = []
 
@@ -352,7 +352,6 @@ def __init__(
         local_ingest_port: int = 14499,
         print_metrics: bool = True,
     ):
-
         self.secrets = {}
         if secrets_path and Path(secrets_path).exists():
             with open(secrets_path) as f:

dynatrace_extension/sdk/extension.py

@@ -9,13 +9,14 @@
 import threading
 import time
 from argparse import ArgumentParser
+from collections.abc import Callable
 from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime, timedelta, timezone
 from enum import Enum
 from itertools import chain
 from pathlib import Path
 from threading import Lock, RLock, active_count
-from typing import Any, Callable, ClassVar, Dict, List, NamedTuple, Optional, Union
+from typing import Any, ClassVar, NamedTuple
 
 from .activation import ActivationConfig, ActivationType
 from .callback import WrappedCallback
@@ -99,7 +100,7 @@ class CountMetricRegistrationEntry(NamedTuple):
     dimensions_list: list[str]
 
     @staticmethod
-    def make_list(metric_key: str, dimensions_list: List[str]):
+    def make_list(metric_key: str, dimensions_list: list[str]):
         """Build an entry that uses defined list of dimensions for aggregation.
 
         Args:
@@ -135,7 +136,7 @@ def registration_items_dict(self):
             return result
 
 
-def _add_sfm_metric(metric: Metric, sfm_metrics: Optional[List[Metric]] = None):
+def _add_sfm_metric(metric: Metric, sfm_metrics: list[Metric] | None = None):
     if sfm_metrics is None:
         sfm_metrics = []
     metric.validate()
@@ -194,24 +195,24 @@ def __init__(self, name: str = "") -> None:
         self._cluster_time_diff: int = 0
 
         # Optional callback to be invoked during the fastcheck
-        self._fast_check_callback: Optional[Callable[[ActivationConfig, str], Status]] = None
+        self._fast_check_callback: Callable[[ActivationConfig, str], Status] | None = None
 
         # List of all scheduled callbacks we must run
-        self._scheduled_callbacks: List[WrappedCallback] = []
-        self._scheduled_callbacks_before_run: List[WrappedCallback] = []
+        self._scheduled_callbacks: list[WrappedCallback] = []
+        self._scheduled_callbacks_before_run: list[WrappedCallback] = []
 
         # Internal callbacks results, used to report statuses
-        self._internal_callbacks_results: Dict[str, Status] = {}
+        self._internal_callbacks_results: dict[str, Status] = {}
         self._internal_callbacks_results_lock: Lock = Lock()
 
         # Running callbacks, used to get the callback info when reporting metrics
-        self._running_callbacks: Dict[int, WrappedCallback] = {}
+        self._running_callbacks: dict[int, WrappedCallback] = {}
         self._running_callbacks_lock: Lock = Lock()
 
         self._scheduler = sched.scheduler(time.time, time.sleep)
 
         # Timestamps for scheduling of internal callbacks
-        self._next_internal_callbacks_timestamps: Dict[str, datetime] = {
+        self._next_internal_callbacks_timestamps: dict[str, datetime] = {
             "timediff": datetime.now() + TIME_DIFF_INTERVAL,
             "heartbeat": datetime.now() + HEARTBEAT_INTERVAL,
             "metrics": datetime.now() + METRIC_SENDING_INTERVAL,
@@ -226,15 +227,15 @@ def __init__(self, name: str = "") -> None:
 
         # Extension metrics
         self._metrics_lock = RLock()
-        self._metrics: List[str] = []
+        self._metrics: list[str] = []
 
         # Extension logs
         self._logs_lock = RLock()
-        self._logs: List[dict] = []
+        self._logs: list[dict] = []
 
         # Self monitoring metrics
         self._sfm_metrics_lock = Lock()
-        self._callbackSfmReport: Dict[str, WrappedCallback] = {}
+        self._callbackSfmReport: dict[str, WrappedCallback] = {}
 
         # Count metric delta signals
         self._delta_signal_buffer: set[str] = set()
@@ -342,9 +343,9 @@ def _schedule_callback(self, callback: WrappedCallback):
     def schedule(
         self,
         callback: Callable,
-        interval: Union[timedelta, int],
-        args: Optional[tuple] = None,
-        activation_type: Optional[ActivationType] = None,
+        interval: timedelta | int,
+        args: tuple | None = None,
+        activation_type: ActivationType | None = None,
     ) -> None:
         """Schedule a method to be executed periodically.
 
@@ -455,11 +456,11 @@ def _send_count_delta_signal(self, metric_keys: set[str], force: bool = True) ->
     def report_metric(
         self,
         key: str,
-        value: Union[float, str, int, SummaryStat],
-        dimensions: Optional[Dict[str, str]] = None,
-        device_address: Optional[str] = None,
-        techrule: Optional[str] = None,
-        timestamp: Optional[datetime] = None,
+        value: float | str | int | SummaryStat,
+        dimensions: dict[str, str] | None = None,
+        device_address: str | None = None,
+        techrule: str | None = None,
+        timestamp: datetime | None = None,
         metric_type: MetricType = MetricType.GAUGE,
     ) -> None:
         """Report a metric.
@@ -498,7 +499,7 @@ def report_metric(
         metric = Metric(key=key, value=value, dimensions=dimensions, metric_type=metric_type, timestamp=timestamp)
         self._add_metric(metric)
 
-    def report_mint_lines(self, lines: List[str]) -> None:
+    def report_mint_lines(self, lines: list[str]) -> None:
         """Report mint lines using the MINT protocol
 
         Examples:
@@ -515,9 +516,9 @@ def report_event(
         self,
         title: str,
         description: str,
-        properties: Optional[dict] = None,
-        timestamp: Optional[datetime] = None,
-        severity: Union[Severity, str] = Severity.INFO,
+        properties: dict | None = None,
+        timestamp: datetime | None = None,
+        severity: Severity | str = Severity.INFO,
         send_immediately: bool = False,
     ) -> None:
         """Report an event using log ingest.
@@ -551,11 +552,11 @@ def report_dt_event(
         self,
         event_type: DtEventType,
         title: str,
-        start_time: Optional[int] = None,
-        end_time: Optional[int] = None,
-        timeout: Optional[int] = None,
-        entity_selector: Optional[str] = None,
-        properties: Optional[dict[str, str]] = None,
+        start_time: int | None = None,
+        end_time: int | None = None,
+        timeout: int | None = None,
+        entity_selector: str | None = None,
+        properties: dict[str, str] | None = None,
     ) -> None:
         """
         Reports an event using the v2 event ingest API.
@@ -575,7 +576,7 @@ def report_dt_event(
             entity_selector: The entity selector, if not set, the event is associated with environment entity (optional)
             properties: A map of event properties (optional)
         """
-        event: Dict[str, Any] = {"eventType": event_type, "title": title}
+        event: dict[str, Any] = {"eventType": event_type, "title": title}
         if start_time:
             event["startTime"] = start_time
         if end_time:
@@ -662,7 +663,7 @@ def report_log_event(self, log_event: dict, send_immediately: bool = False):
         """
         self._send_events(log_event, send_immediately=send_immediately)
 
-    def report_log_events(self, log_events: List[dict], send_immediately: bool = False):
+    def report_log_events(self, log_events: list[dict], send_immediately: bool = False):
         """Report a list of custom log events using log ingest.
 
         Args:
@@ -671,7 +672,7 @@ def report_log_events(self, log_events: List[dict], send_immediately: bool = Fal
         """
         self._send_events(log_events, send_immediately=send_immediately)
 
-    def report_log_lines(self, log_lines: List[Union[str, bytes]], send_immediately: bool = False):
+    def report_log_lines(self, log_lines: list[str | bytes], send_immediately: bool = False):
         """Report a list of log lines using log ingest
 
         Args:
@@ -885,13 +886,13 @@ def _send_metrics(self):
                     api_logger.info(f"Sent {number_of_metrics} metric lines to EEC: {responses}")
                     self._metrics = []
 
-    def _prepare_sfm_metrics(self) -> List[str]:
+    def _prepare_sfm_metrics(self) -> list[str]:
         """Prepare self monitoring metrics.
 
         Builds the list of mint metric lines to send as self monitoring metrics.
         """
 
-        sfm_metrics: List[Metric] = []
+        sfm_metrics: list[Metric] = []
         sfm_dimensions = {"dt.extension.config.id": self.monitoring_config_id}
         _add_sfm_metric(
             SfmMetric("threads", active_count(), sfm_dimensions, client_facing=True, metric_type=MetricType.DELTA),
@@ -1047,11 +1048,11 @@ def _add_metric(self, metric: Metric):
         with self._metrics_lock:
             self._metrics.append(metric.to_mint_line())
 
-    def _add_mint_lines(self, lines: List[str]):
+    def _add_mint_lines(self, lines: list[str]):
         with self._metrics_lock:
             self._metrics.extend(lines)
 
-    def _send_events_internal(self, events: Union[dict, List[dict]]):
+    def _send_events_internal(self, events: dict | list[dict]):
         try:
             responses = self._client.send_events(events, self.log_event_enrichment)
 
@@ -1068,7 +1069,7 @@ def _send_events_internal(self, events: Union[dict, List[dict]]):
             with self._internal_callbacks_results_lock:
                 self._internal_callbacks_results[self._send_events.__name__] = Status(StatusValue.GENERIC_ERROR, str(e))
 
-    def _send_events(self, events: Union[dict, List[dict]], send_immediately: bool = False):
+    def _send_events(self, events: dict | list[dict], send_immediately: bool = False):
         if send_immediately:
             self._internal_executor.submit(self._send_events_internal, events)
             return