Merge pull request #3 from UiPath/fix/typing

fix: mypy errors

Author: cristipufu

pyproject.toml

@@ -76,7 +76,9 @@ line-ending = "auto"
 
 [tool.mypy]
 plugins = ["pydantic.mypy"]
-
+mypy_path = "src"
+explicit_package_bases = true
+namespace_packages = true
 follow_imports = "silent"
 warn_redundant_casts = true
 warn_unused_ignores = true

src/uipath/runtime/logging/_filters.py

@@ -24,7 +24,7 @@ def filter(self, record: logging.LogRecord) -> bool:
         ctx_execution_id = current_execution_id.get()
         if ctx_execution_id == self.execution_id:
             # Inject execution_id into record for downstream handlers
-            record.execution_id = self.execution_id  # type: ignore[attr-defined]
+            record.execution_id = self.execution_id
             return True
 
         return False

src/uipath/runtime/logging/_interceptor.py

@@ -170,10 +170,10 @@ def _redirect_stdout_stderr(self) -> None:
             self._clean_all_handlers(stderr_logger)
 
         # Use the min_level in the LoggerWriter to filter messages
-        sys.stdout = LoggerWriter(  # type: ignore[assignment]
+        sys.stdout = LoggerWriter(
             stdout_logger, logging.INFO, self.numeric_min_level, self.original_stdout
         )
-        sys.stderr = LoggerWriter(  # type: ignore[assignment]
+        sys.stderr = LoggerWriter(
             stderr_logger, logging.ERROR, self.numeric_min_level, self.original_stderr
         )
 

py.typed src/uipath/runtime/py.typedpy.typed renamed to src/uipath/runtime/py.typed

@@ -5,11 +5,12 @@
 import os
 import random
 import uuid
+from collections.abc import Callable
 from dataclasses import asdict, is_dataclass
 from datetime import datetime, timezone
 from enum import Enum
 from os import environ as env
-from typing import Any, Callable, Dict, Optional
+from typing import Any, Dict, Mapping, Optional
 from zoneinfo import ZoneInfo
 
 from opentelemetry.sdk.trace import ReadableSpan
@@ -20,35 +21,21 @@
 
 
 def get_supported_params(
-    tracer_impl: Callable, params: Dict[str, Any]
+    tracer_impl: Callable[..., Any],
+    params: Mapping[str, Any],
 ) -> Dict[str, Any]:
-    """Extract the parameters supported by the tracer implementation.
-
-    Args:
-        tracer_impl: The tracer implementation function or callable
-        params: Dictionary of parameters to check
-
-    Returns:
-        Dictionary containing only parameters supported by the tracer implementation
-    """
-    supported_params = {}
-    if hasattr(tracer_impl, "__code__"):
-        # For regular functions
-        impl_signature = inspect.signature(tracer_impl)
-        for param_name, param_value in params.items():
-            if param_name in impl_signature.parameters and param_value is not None:
-                supported_params[param_name] = param_value
-    elif callable(tracer_impl):
-        # For callable objects
-        impl_signature = inspect.signature(tracer_impl.__call__)
-        for param_name, param_value in params.items():
-            if param_name in impl_signature.parameters and param_value is not None:
-                supported_params[param_name] = param_value
-    else:
+    """Extract the parameters supported by the tracer implementation."""
+    try:
+        sig = inspect.signature(tracer_impl)
+    except (TypeError, ValueError):
         # If we can't inspect, pass all parameters and let the function handle it
-        supported_params = params
+        return dict(params)
 
-    return supported_params
+    supported: Dict[str, Any] = {}
+    for name, value in params.items():
+        if value is not None and name in sig.parameters:
+            supported[name] = value
+    return supported
 
 
 def _simple_serialize_defaults(obj):

src/uipath/runtime/tracing/_utils.py

@@ -19,12 +19,12 @@
 logger = logging.getLogger(__name__)
 
 
-def _default_input_processor() -> dict:
+def _default_input_processor() -> dict[str, Any]:
     """Default input processor that doesn't log any actual input data."""
     return {"redacted": "Input data not logged for privacy/security"}
 
 
-def _default_output_processor() -> dict:
+def _default_output_processor() -> dict[str, Any]:
     """Default output processor that doesn't log any actual output data."""
     return {"redacted": "Output data not logged for privacy/security"}
 

src/uipath/runtime/tracing/decorators.py

@@ -1,6 +1,6 @@
 """Custom span processors for UiPath execution tracing."""
 
-from typing import Optional
+from typing import Optional, cast
 
 from opentelemetry import context as context_api
 from opentelemetry import trace
@@ -18,11 +18,11 @@ def on_start(
         """Called when a span is started."""
         parent_span: Optional[Span]
         if parent_context:
-            parent_span = trace.get_current_span(parent_context)
+            parent_span = cast(Span, trace.get_current_span(parent_context))
         else:
-            parent_span = trace.get_current_span()
+            parent_span = cast(Span, trace.get_current_span())
 
-        if parent_span and parent_span.is_recording():
+        if parent_span and parent_span.is_recording() and parent_span.attributes:
             execution_id = parent_span.attributes.get("execution.id")
             if execution_id:
                 span.set_attribute("execution.id", execution_id)

src/uipath/runtime/tracing/processors.py

@@ -2,7 +2,7 @@
 
 from datetime import datetime
 from os import environ as env
-from typing import Optional
+from typing import Any, Optional
 from uuid import UUID
 
 from pydantic import BaseModel, ConfigDict, Field
@@ -62,7 +62,7 @@ class UiPathRuntimeSpan(BaseModel):
         default_factory=lambda: env.get("UIPATH_JOB_KEY"), serialization_alias="JobKey"
     )
 
-    def to_dict(self) -> dict:
+    def to_dict(self) -> dict[str, Any]:
         """Convert the Span to a dictionary suitable for JSON serialization.
 
         Returns a dict with PascalCase keys for UiPath API compatibility.