chore(profiling): improve typing in scheduler.py

Author: KowalskiThomas

ddtrace/profiling/scheduler.py

@@ -1,7 +1,7 @@
 # -*- encoding: utf-8 -*-
 import logging
 import time
-from typing import Any  # noqa F401
+from typing import Any
 from typing import Callable
 from typing import Optional
 
@@ -20,27 +20,25 @@ class Scheduler(periodic.PeriodicService):
 
     def __init__(
         self,
-        before_flush: Optional[Callable] = None,
+        before_flush: Optional[Callable[[], None]] = None,
         tracer: Optional[Tracer] = ddtrace.tracer,
         interval: float = config.upload_interval,
-    ):
+    ) -> None:
         super(Scheduler, self).__init__(interval=interval)
-        self.before_flush: Optional[Callable] = before_flush
+        self.before_flush: Optional[Callable[[], None]] = before_flush
         self._configured_interval: float = self.interval
         self._last_export: int = 0  # Overridden in _start_service
-        self._tracer = tracer
+        self._tracer: Optional[Tracer] = tracer
         self._enable_code_provenance: bool = config.code_provenance
 
-    def _start_service(self):
-        # type: (...) -> None
+    def _start_service(self) -> None:
         """Start the scheduler."""
         LOG.debug("Starting scheduler")
         super(Scheduler, self)._start_service()
         self._last_export = time.time_ns()
         LOG.debug("Scheduler started")
 
-    def flush(self):
-        # type: (...) -> None
+    def flush(self) -> None:
         """Flush events from recorder to exporters."""
         LOG.debug("Flushing events")
         if self.before_flush is not None:
@@ -53,8 +51,7 @@ def flush(self):
 
         self._last_export = time.time_ns()
 
-    def periodic(self):
-        # type: (...) -> None
+    def periodic(self) -> None:
         start_time = time.monotonic()
         try:
             self.flush()
@@ -67,22 +64,20 @@ class ServerlessScheduler(Scheduler):
 
     The idea with this scheduler is to not sleep 60s, but to sleep 1s and flush out profiles after 60 sleeping period.
     As the service can be frozen a few seconds after flushing out a profile, we want to make sure the next flush is not
-    > 60s later, but after at least 60 periods of 1s.
+    > 60s later, but after at least 60 periods of 1s.
 
     """
 
     # We force this interval everywhere
     FORCED_INTERVAL = 1.0
     FLUSH_AFTER_INTERVALS = 60.0
 
-    def __init__(self, *args, **kwargs):
-        # type: (*Any, **Any) -> None
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
         kwargs.setdefault("interval", self.FORCED_INTERVAL)
         super(ServerlessScheduler, self).__init__(*args, **kwargs)
         self._profiled_intervals: int = 0
 
-    def periodic(self):
-        # type: (...) -> None
+    def periodic(self) -> None:
         # Check both the number of intervals and time frame to be sure we don't flush, e.g., empty profiles
         if self._profiled_intervals >= self.FLUSH_AFTER_INTERVALS and (time.time_ns() - self._last_export) >= (
             self.FORCED_INTERVAL * self.FLUSH_AFTER_INTERVALS