dummy

Signed-off-by: Cagri Yonca <cagri@ibm.com>

Author: CagriYonca

src/instana/options.py

@@ -14,9 +14,12 @@
     - GCROptions - Options class for Google cloud Run.  Holds settings specific to GCR.
 """
 
+from __future__ import annotations
+
 import logging
 import os
-from typing import Any, Dict, Sequence, Tuple
+from collections.abc import Sequence
+from typing import Any
 
 from instana.configurator import config
 from instana.log import logger
@@ -41,7 +44,7 @@
 class BaseOptions(object):
     """Base class for all option classes.  Holds items common to all"""
 
-    def __init__(self, **kwds: Dict[str, Any]) -> None:
+    def __init__(self, **kwds: object) -> None:
         self.debug = False
         self.log_level = logging.WARN
         self.service_name = determine_service_name()
@@ -324,7 +327,7 @@ def is_span_disabled(self, category=None, span_type=None) -> bool:
         # Default: not disabled
         return False
 
-    def get_stack_trace_config(self, span_name: str) -> Tuple[str, int]:
+    def get_stack_trace_config(self, span_name: str) -> tuple[str, int]:
         """
         Get stack trace configuration for a specific span type.
         Technology-specific configuration overrides global configuration.
@@ -362,7 +365,7 @@ class StandardOptions(BaseOptions):
     DEFAULT_POLL_RATE = 1
     MAX_POLL_RATE = 5
 
-    def __init__(self, **kwds: Dict[str, Any]) -> None:
+    def __init__(self, **kwds: object) -> None:
         super(StandardOptions, self).__init__()
 
         self.agent_host = os.environ.get("INSTANA_AGENT_HOST", self.AGENT_DEFAULT_HOST)
@@ -372,7 +375,7 @@ def __init__(self, **kwds: Dict[str, Any]) -> None:
         if not isinstance(self.agent_port, int):
             self.agent_port = int(self.agent_port)
 
-    def set_secrets(self, secrets: Dict[str, Any]) -> None:
+    def set_secrets(self, secrets: dict[str, str | list[str]]) -> None:
         """
         Set the secret option from the agent config.
         @param secrets: dictionary of secrets
@@ -381,7 +384,7 @@ def set_secrets(self, secrets: Dict[str, Any]) -> None:
         self.secrets_matcher = secrets["matcher"]
         self.secrets_list = secrets["list"]
 
-    def set_extra_headers(self, extra_headers: Dict[str, Any]) -> None:
+    def set_extra_headers(self, extra_headers: list[str]) -> None:
         """
         Set the extra headers option from the agent config, which uses the legacy configuration setting.
         @param extra_headers: dictionary of headers
@@ -395,7 +398,7 @@ def set_extra_headers(self, extra_headers: Dict[str, Any]) -> None:
             f"Will also capture these custom headers: {self.extra_http_headers}"
         )
 
-    def set_tracing(self, tracing: Dict[str, Any]) -> None:
+    def set_tracing(self, tracing: dict[str, Any]) -> None:
         """
         Set tracing options from the agent config.
         @param tracing: tracing configuration dictionary
@@ -417,7 +420,7 @@ def set_tracing(self, tracing: Dict[str, Any]) -> None:
         # Handle stack trace configuration from agent config
         self.set_stack_trace_from_agent(tracing)
 
-    def _apply_agent_filter_config(self, filter_config: Any) -> None:
+    def _apply_agent_filter_config(self, filter_config: dict[str, Any]) -> None:
         """Apply span filter rules from agent config."""
         parsed = parse_filter_rules(filter_config)
         for policy in ("exclude", "include"):
@@ -427,7 +430,7 @@ def _apply_agent_filter_config(self, filter_config: Any) -> None:
                     self.span_filters[policy] = []
                 self.span_filters[policy].extend(rules)
 
-    def _apply_agent_kafka_config(self, kafka_config: Dict[str, Any]) -> None:
+    def _apply_agent_kafka_config(self, kafka_config: dict[str, str | bool]) -> None:
         """Apply Kafka tracing configuration from agent config."""
         no_env_override = "INSTANA_KAFKA_TRACE_CORRELATION" not in os.environ
         no_code_override = not (
@@ -476,7 +479,7 @@ def _should_apply_agent_global_config(self) -> bool:
         return not (has_env_vars or has_yaml_config or has_in_code_config)
 
     def _apply_agent_global_stack_trace_config(
-        self, global_config: Dict[str, Any]
+        self, global_config: dict[str, Any]
     ) -> None:
         """Apply global stack trace configuration from agent config."""
         if "stack-trace" in global_config and (
@@ -493,7 +496,7 @@ def _apply_agent_global_stack_trace_config(
         ):
             self.stack_trace_length = validated_length
 
-    def _apply_agent_tech_stack_trace_config(self, tracing: Dict[str, Any]) -> None:
+    def _apply_agent_tech_stack_trace_config(self, tracing: dict[str, Any]) -> None:
         """Apply technology-specific stack trace configuration from agent config."""
         for tech_name, tech_config in tracing.items():
             if tech_name == "global" or not isinstance(tech_config, dict):
@@ -509,7 +512,7 @@ def _apply_agent_tech_stack_trace_config(self, tracing: Dict[str, Any]) -> None:
             if tech_stack_config:
                 self.stack_trace_technology_config[tech_name] = tech_stack_config
 
-    def set_stack_trace_from_agent(self, tracing: Dict[str, Any]) -> None:
+    def set_stack_trace_from_agent(self, tracing: dict[str, Any]) -> None:
         """
         Set stack trace configuration from agent config (configuration.yaml).
         Only applies if not already set by higher priority sources.
@@ -524,7 +527,7 @@ def set_stack_trace_from_agent(self, tracing: Dict[str, Any]) -> None:
         if not self.stack_trace_technology_config:
             self._apply_agent_tech_stack_trace_config(tracing)
 
-    def set_disable_tracing(self, tracing_config: Sequence[Dict[str, Any]]) -> None:
+    def set_disable_tracing(self, tracing_config: Sequence[dict[str, Any]]) -> None:
         # The precedence is as follows:
         # environment variables > in-code (local) config > agent config (configuration.yaml)
         if (
@@ -540,7 +543,7 @@ def set_disable_tracing(self, tracing_config: Sequence[Dict[str, Any]]) -> None:
             self.disabled_spans.extend(disabled_spans)
             self.enabled_spans.extend(enabled_spans)
 
-    def set_poll_rate(self, plugin_config: Dict[str, Any]) -> None:
+    def set_poll_rate(self, plugin_config: dict[str, Any]) -> None:
         """Set poll rate from agent plugin configuration."""
         poll_rate_value = plugin_config.get("poll_rate")
         if poll_rate_value is None:
@@ -568,7 +571,7 @@ def set_poll_rate(self, plugin_config: Dict[str, Any]) -> None:
         )
         self.poll_rate = self.DEFAULT_POLL_RATE
 
-    def set_from(self, res_data: Dict[str, Any]) -> None:
+    def set_from(self, res_data: dict[str, Any]) -> None:
         """
         Set the source identifiers given to use by the Instana Host agent.
         @param res_data: source identifiers provided as announce response
@@ -598,7 +601,7 @@ def set_from(self, res_data: Dict[str, Any]) -> None:
 class ServerlessOptions(BaseOptions):
     """Base class for serverless environments.  Holds settings common to all serverless environments."""
 
-    def __init__(self, **kwds: Dict[str, Any]) -> None:
+    def __init__(self, **kwds: object) -> None:
         super(ServerlessOptions, self).__init__()
 
         self.agent_key = os.environ.get("INSTANA_AGENT_KEY", None)
@@ -657,14 +660,14 @@ def __init__(self, **kwds: Dict[str, Any]) -> None:
 class AWSLambdaOptions(ServerlessOptions):
     """Options class for AWS Lambda.  Holds settings specific to AWS Lambda."""
 
-    def __init__(self, **kwds: Dict[str, Any]) -> None:
+    def __init__(self, **kwds: object) -> None:
         super(AWSLambdaOptions, self).__init__()
 
 
 class AWSFargateOptions(ServerlessOptions):
     """Options class for AWS Fargate.  Holds settings specific to AWS Fargate."""
 
-    def __init__(self, **kwds: Dict[str, Any]) -> None:
+    def __init__(self, **kwds: object) -> None:
         super(AWSFargateOptions, self).__init__()
 
         self.tags = None
@@ -689,12 +692,12 @@ def __init__(self, **kwds: Dict[str, Any]) -> None:
 class EKSFargateOptions(AWSFargateOptions):
     """Options class for EKS Pods on AWS Fargate. Holds settings specific to EKS Pods on AWS Fargate."""
 
-    def __init__(self, **kwds: Dict[str, Any]) -> None:
+    def __init__(self, **kwds: object) -> None:
         super(EKSFargateOptions, self).__init__()
 
 
 class GCROptions(ServerlessOptions):
     """Options class for Google Cloud Run.  Holds settings specific to Google Cloud Run."""
 
-    def __init__(self, **kwds: Dict[str, Any]) -> None:
+    def __init__(self, **kwds: object) -> None:
         super(GCROptions, self).__init__()