chore: add type hinting to tornado instrumentation

Signed-off-by: Cagri Yonca <[email protected]>

Author: CagriYonca

src/instana/instrumentation/tornado/client.py

@@ -1,30 +1,41 @@
-# (c) Copyright IBM Corp. 2021
+# (c) Copyright IBM Corp. 2025
 # (c) Copyright Instana Inc. 2019
 
 try:
     import tornado
 
     import wrapt
     import functools
-    from typing import TYPE_CHECKING, Dict, Any
+    from typing import TYPE_CHECKING, Dict, Any, Callable, Tuple
+
+    if TYPE_CHECKING:
+        from instana.span.span import InstanaSpan
+        from asyncio import Future
+        from tornado.httpclient import AsyncHTTPClient
 
     from opentelemetry.semconv.trace import SpanAttributes
 
     from instana.log import logger
-    from instana.singletons import agent, tracer
+    from instana.singletons import agent, get_tracer
     from instana.util.secrets import strip_secrets_from_query
     from instana.util.traceutils import extract_custom_headers
     from instana.propagators.format import Format
     from instana.span.span import get_current_span
 
-
-    @wrapt.patch_function_wrapper('tornado.httpclient', 'AsyncHTTPClient.fetch')
-    def fetch_with_instana(wrapped, instance, argv, kwargs):
+    @wrapt.patch_function_wrapper("tornado.httpclient", "AsyncHTTPClient.fetch")
+    def fetch_with_instana(
+        wrapped: Callable[..., object],
+        instance: "AsyncHTTPClient",
+        argv: Tuple[object, ...],
+        kwargs: Dict[str, Any],
+    ) -> "Future":
         try:
             parent_span = get_current_span()
 
             # If we're not tracing, just return
-            if (not parent_span.is_recording()) or (parent_span.name == "tornado-client"):
+            if (not parent_span.is_recording()) or (
+                parent_span.name == "tornado-client"
+            ):
                 return wrapped(*argv, **kwargs)
 
             request = argv[0]
@@ -35,25 +46,26 @@ def fetch_with_instana(wrapped, instance, argv, kwargs):
                 request = tornado.httpclient.HTTPRequest(url=request, **kwargs)
 
                 new_kwargs = {}
-                for param in ('callback', 'raise_error'):
+                for param in ("callback", "raise_error"):
                     # if not in instead and pop
                     if param in kwargs:
                         new_kwargs[param] = kwargs.pop(param)
                 kwargs = new_kwargs
 
             parent_context = parent_span.get_span_context() if parent_span else None
-
+            tracer = get_tracer()
             span = tracer.start_span("tornado-client", span_context=parent_context)
 
             extract_custom_headers(span, request.headers)
 
             tracer.inject(span.context, Format.HTTP_HEADERS, request.headers)
 
             # Query param scrubbing
-            parts = request.url.split('?')
+            parts = request.url.split("?")
             if len(parts) > 1:
-                cleaned_qp = strip_secrets_from_query(parts[1], agent.options.secrets_matcher,
-                                                      agent.options.secrets_list)
+                cleaned_qp = strip_secrets_from_query(
+                    parts[1], agent.options.secrets_matcher, agent.options.secrets_list
+                )
                 span.set_attribute("http.params", cleaned_qp)
 
             span.set_attribute(SpanAttributes.HTTP_URL, parts[0])
@@ -69,8 +81,7 @@ def fetch_with_instana(wrapped, instance, argv, kwargs):
         except Exception:
             logger.debug("Tornado fetch_with_instana: ", exc_info=True)
 
-
-    def finish_tracing(future, span):
+    def finish_tracing(future: "Future", span: "InstanaSpan") -> None:
         try:
             response = future.result()
             span.set_attribute(SpanAttributes.HTTP_STATUS_CODE, response.code)
@@ -84,7 +95,6 @@ def finish_tracing(future, span):
             if span.is_recording():
                 span.end()
 
-
     logger.debug("Instrumenting tornado client")
 except ImportError:
     pass

src/instana/instrumentation/tornado/server.py

@@ -1,38 +1,53 @@
-# (c) Copyright IBM Corp. 2021
+# (c) Copyright IBM Corp. 2025
 # (c) Copyright Instana Inc. 2019
 
 
 try:
     import tornado
+    from typing import TYPE_CHECKING, Callable, Tuple, Dict, Any, Coroutine, Optional
 
     import wrapt
 
+    if TYPE_CHECKING:
+        from tornado.web import RequestHandler
+
     from opentelemetry.semconv.trace import SpanAttributes
 
     from instana.log import logger
-    from instana.singletons import agent, tracer
+    from instana.singletons import agent, get_tracer
     from instana.util.secrets import strip_secrets_from_query
     from instana.util.traceutils import extract_custom_headers
     from instana.propagators.format import Format
 
-
-
-    @wrapt.patch_function_wrapper('tornado.web', 'RequestHandler._execute')
-    def execute_with_instana(wrapped, instance, argv, kwargs):
+    @wrapt.patch_function_wrapper("tornado.web", "RequestHandler._execute")
+    def execute_with_instana(
+        wrapped: Callable[..., object],
+        instance: "RequestHandler",
+        argv: Tuple[object, ...],
+        kwargs: Dict[str, Any],
+    ) -> Coroutine:
         try:
             span_context = None
-            if hasattr(instance.request.headers, '__dict__') and '_dict' in instance.request.headers.__dict__:
-                span_context = tracer.extract(Format.HTTP_HEADERS,
-                                                instance.request.headers.__dict__['_dict'])
+            tracer = get_tracer()
+            if (
+                hasattr(instance.request.headers, "__dict__")
+                and "_dict" in instance.request.headers.__dict__
+            ):
+                span_context = tracer.extract(
+                    Format.HTTP_HEADERS, instance.request.headers.__dict__["_dict"]
+                )
 
             span = tracer.start_span("tornado-server", span_context=span_context)
 
             # Query param scrubbing
             if instance.request.query is not None and len(instance.request.query) > 0:
-                cleaned_qp = strip_secrets_from_query(instance.request.query, agent.options.secrets_matcher,
-                                                        agent.options.secrets_list)
+                cleaned_qp = strip_secrets_from_query(
+                    instance.request.query,
+                    agent.options.secrets_matcher,
+                    agent.options.secrets_list,
+                )
                 span.set_attribute("http.params", cleaned_qp)
-            
+
             url = f"{instance.request.protocol}://{instance.request.host}{instance.request.path}"
             span.set_attribute(SpanAttributes.HTTP_URL, url)
             span.set_attribute(SpanAttributes.HTTP_METHOD, instance.request.method)
@@ -52,20 +67,29 @@ def execute_with_instana(wrapped, instance, argv, kwargs):
         except Exception:
             logger.debug("tornado execute", exc_info=True)
 
-
-    @wrapt.patch_function_wrapper('tornado.web', 'RequestHandler.set_default_headers')
-    def set_default_headers_with_instana(wrapped, instance, argv, kwargs):
-        if not hasattr(instance.request, '_instana'):
+    @wrapt.patch_function_wrapper("tornado.web", "RequestHandler.set_default_headers")
+    def set_default_headers_with_instana(
+        wrapped: Callable[..., object],
+        instance: "RequestHandler",
+        argv: Tuple[object, ...],
+        kwargs: Dict[str, Any],
+    ) -> Optional[Coroutine]:
+        if not hasattr(instance.request, "_instana"):
             return wrapped(*argv, **kwargs)
 
         span = instance.request._instana
+        tracer = get_tracer()
         tracer.inject(span.context, Format.HTTP_HEADERS, instance._headers)
 
-
-    @wrapt.patch_function_wrapper('tornado.web', 'RequestHandler.on_finish')
-    def on_finish_with_instana(wrapped, instance, argv, kwargs):
+    @wrapt.patch_function_wrapper("tornado.web", "RequestHandler.on_finish")
+    def on_finish_with_instana(
+        wrapped: Callable[..., object],
+        instance: "RequestHandler",
+        argv: Tuple[object, ...],
+        kwargs: Dict[str, Any],
+    ) -> Coroutine:
         try:
-            if not hasattr(instance.request, '_instana'):
+            if not hasattr(instance.request, "_instana"):
                 return wrapped(*argv, **kwargs)
 
             span = instance.request._instana
@@ -86,11 +110,15 @@ def on_finish_with_instana(wrapped, instance, argv, kwargs):
         except Exception:
             logger.debug("tornado on_finish", exc_info=True)
 
-
-    @wrapt.patch_function_wrapper('tornado.web', 'RequestHandler.log_exception')
-    def log_exception_with_instana(wrapped, instance, argv, kwargs):
+    @wrapt.patch_function_wrapper("tornado.web", "RequestHandler.log_exception")
+    def log_exception_with_instana(
+        wrapped: Callable[..., object],
+        instance: "RequestHandler",
+        argv: Tuple[object, ...],
+        kwargs: Dict[str, Any],
+    ) -> Coroutine:
         try:
-            if not hasattr(instance.request, '_instana'):
+            if not hasattr(instance.request, "_instana"):
                 return wrapped(*argv, **kwargs)
 
             if not isinstance(argv[1], tornado.web.HTTPError):
@@ -101,7 +129,6 @@ def log_exception_with_instana(wrapped, instance, argv, kwargs):
         except Exception:
             logger.debug("tornado log_exception", exc_info=True)
 
-
     logger.debug("Instrumenting tornado server")
 except ImportError:
     pass