Admin: Add typehints to utils/strings and utils/threads

bblommers · bblommers · commit b66207d6002a · 2026-01-29T06:30:16.000-01:00
diff --git a/localstack-core/localstack/utils/run.py b/localstack-core/localstack/utils/run.py
@@ -170,7 +170,7 @@ def is_command_available(cmd: str) -> bool:
         return False
 
 
-def kill_process_tree(parent_pid):
+def kill_process_tree(parent_pid: int) -> None:
     # Note: Do NOT import "psutil" at the root scope
     import psutil
 
diff --git a/localstack-core/localstack/utils/strings.py b/localstack-core/localstack/utils/strings.py
@@ -7,9 +7,13 @@
 import string
 import uuid
 import zlib
+from typing import TYPE_CHECKING, Any
 
 from localstack.config import DEFAULT_ENCODING
 
+if TYPE_CHECKING:
+    from localstack.utils.objects import ComplexType
+
 _unprintables = (
     range(0x00, 0x09),
     range(0x0A, 0x0A),
@@ -27,13 +31,13 @@
 )
 
 
-def to_str(obj: str | bytes, encoding: str = DEFAULT_ENCODING, errors="strict") -> str:
+def to_str(obj: str | bytes, encoding: str = DEFAULT_ENCODING, errors: str = "strict") -> str:
     """If ``obj`` is an instance of ``binary_type``, return
     ``obj.decode(encoding, errors)``, otherwise return ``obj``"""
     return obj.decode(encoding, errors) if isinstance(obj, bytes) else obj
 
 
-def to_bytes(obj: str | bytes, encoding: str = DEFAULT_ENCODING, errors="strict") -> bytes:
+def to_bytes(obj: str | bytes, encoding: str = DEFAULT_ENCODING, errors: str = "strict") -> bytes:
     """If ``obj`` is an instance of ``text_type``, return
     ``obj.encode(encoding, errors)``, otherwise return ``obj``"""
     return obj.encode(encoding, errors) if isinstance(obj, str) else obj
@@ -44,7 +48,7 @@ def truncate(data: str, max_length: int = 100) -> str:
     return (f"{data[:max_length]}...") if len(data) > max_length else data
 
 
-def is_string(s, include_unicode=True, exclude_binary=False):
+def is_string(s: Any, include_unicode: bool = True, exclude_binary: bool = False) -> bool:
     if isinstance(s, bytes) and exclude_binary:
         return False
     if isinstance(s, str):
@@ -54,13 +58,13 @@ def is_string(s, include_unicode=True, exclude_binary=False):
     return False
 
 
-def is_string_or_bytes(s):
+def is_string_or_bytes(s: Any) -> bool:
     return is_string(s) or isinstance(s, str) or isinstance(s, bytes)
 
 
-def is_base64(s):
+def is_base64(s: Any) -> bool:
     regex = r"^(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?$"
-    return is_string(s) and re.match(regex, s)
+    return is_string(s) and re.match(regex, s) is not None
 
 
 _re_camel_to_snake_case = re.compile("((?<=[a-z0-9])[A-Z]|(?!^)[A-Z](?=[a-z]))")
@@ -85,22 +89,21 @@ def canonicalize_bool_to_str(val: bool) -> str:
     return "true" if str(val).lower() == "true" else "false"
 
 
-def convert_to_printable_chars(value: list | dict | str) -> str:
+def convert_to_printable_chars(value: "str | ComplexType") -> "ComplexType":
     """Removes all unprintable characters from the given string."""
     from localstack.utils.objects import recurse_object
 
-    if isinstance(value, (dict, list)):
+    if isinstance(value, str):
+        return REGEX_UNPRINTABLE_CHARS.sub("", value)
+    else:
 
-        def _convert(obj, **kwargs):
+        def _convert(obj: Any, **kwargs: Any) -> "ComplexType":
             if isinstance(obj, str):
                 return convert_to_printable_chars(obj)
             return obj
 
         return recurse_object(value, _convert)
 
-    result = REGEX_UNPRINTABLE_CHARS.sub("", value)
-    return result
-
 
 def first_char_to_lower(s: str) -> str:
     return s and f"{s[0].lower()}{s[1:]}"
@@ -110,20 +113,20 @@ def first_char_to_upper(s: str) -> str:
     return s and f"{s[0].upper()}{s[1:]}"
 
 
-def str_to_bool(value):
+def str_to_bool(value: Any) -> Any:
     """Return the boolean value of the given string, or the verbatim value if it is not a string"""
     if isinstance(value, str):
         true_strings = ["true", "True"]
         return value in true_strings
     return value
 
 
-def str_insert(string, index, content):
+def str_insert(string: str, index: int, content: str) -> str:
     """Insert a substring into an existing string at a certain index."""
     return f"{string[:index]}{content}{string[index:]}"
 
 
-def str_remove(string, index, end_index=None):
+def str_remove(string: str, index: int, end_index: int | None = None) -> str:
     """Remove a substring from an existing string at a certain from-to index range."""
     end_index = end_index or (index + 1)
     return f"{string[:index]}{string[end_index:]}"
@@ -159,7 +162,7 @@ def checksum_crc32(string: str | bytes) -> str:
     return base64.b64encode(checksum.to_bytes(4, "big")).decode()
 
 
-def checksum_crc32c(string: str | bytes):
+def checksum_crc32c(string: str | bytes) -> str:
     # import botocore locally here to avoid a dependency of the CLI to botocore
     from botocore.httpchecksum import CrtCrc32cChecksum
 
@@ -168,7 +171,7 @@ def checksum_crc32c(string: str | bytes):
     return base64.b64encode(checksum.digest()).decode()
 
 
-def checksum_crc64nvme(string: str | bytes):
+def checksum_crc64nvme(string: str | bytes) -> str:
     # import botocore locally here to avoid a dependency of the CLI to botocore
     from botocore.httpchecksum import CrtCrc64NvmeChecksum
 
@@ -223,7 +226,9 @@ def prepend_with_slash(input: str) -> str:
     return input
 
 
-def key_value_pairs_to_dict(pairs: str, delimiter: str = ",", separator: str = "=") -> dict:
+def key_value_pairs_to_dict(
+    pairs: str, delimiter: str = ",", separator: str = "="
+) -> dict[str, str]:
     """
     Converts a string of key-value pairs to a dictionary.
 
diff --git a/localstack-core/localstack/utils/threads.py b/localstack-core/localstack/utils/threads.py
@@ -1,17 +1,25 @@
 import concurrent.futures
-import inspect
 import logging
+import subprocess
 import threading
 import traceback
 from collections.abc import Callable
 from concurrent.futures import Future
 from multiprocessing.dummy import Pool
+from typing import TYPE_CHECKING, Any, TypeVar
+
+if TYPE_CHECKING:
+    from typing_extensions import ParamSpec
+
+    P = ParamSpec("P")
+
+T = TypeVar("T")
 
 LOG = logging.getLogger(__name__)
 
 # arrays for temporary threads and resources
-TMP_THREADS = []
-TMP_PROCESSES = []
+TMP_THREADS: list["FuncThread"] = []
+TMP_PROCESSES: list[subprocess.Popen[Any]] = []
 
 counter_lock = threading.Lock()
 counter = 0
@@ -22,12 +30,12 @@ class FuncThread(threading.Thread):
 
     def __init__(
         self,
-        func,
-        params=None,
-        quiet=False,
-        on_stop: Callable[["FuncThread"], None] = None,
+        func: "Callable[P, T]",
+        params: Any = None,
+        quiet: bool = False,
+        on_stop: Callable[["FuncThread"], None] | None = None,
         name: str | None = None,
-        daemon=True,
+        daemon: bool = True,
     ):
         global counter
         global counter_lock
@@ -46,17 +54,14 @@ def __init__(
         self.params = params
         self.func = func
         self.quiet = quiet
-        self.result_future = Future()
+        self.result_future: Future[T | Exception | None] = Future()
         self._stop_event = threading.Event()
         self.on_stop = on_stop
 
-    def run(self):
-        result = None
+    def run(self) -> None:
+        result: Any = None
         try:
-            kwargs = {}
-            argspec = inspect.getfullargspec(self.func)
-            if argspec.varkw or "_thread" in (argspec.args or []) + (argspec.kwonlyargs or []):
-                kwargs["_thread"] = self
+            kwargs = {}  # type: ignore[var-annotated]
             result = self.func(self.params, **kwargs)
         except Exception as e:
             self.result_future.set_exception(e)
@@ -78,7 +83,7 @@ def run(self):
                 LOG.debug(e)
 
     @property
-    def running(self):
+    def running(self) -> bool:
         return not self._stop_event.is_set()
 
     def stop(self, quiet: bool = False) -> None:
@@ -91,27 +96,33 @@ def stop(self, quiet: bool = False) -> None:
                 LOG.warning("error while calling on_stop callback: %s", e)
 
 
-def start_thread(method, *args, **kwargs) -> FuncThread:  # TODO: find all usages and add names...
+def start_thread(
+    method: "Callable[P, T]",
+    params: Any = None,
+    quiet: bool = False,
+    on_stop: Callable[["FuncThread"], None] | None = None,
+    _shutdown_hook: bool = True,
+    name: str | None = None,
+) -> FuncThread:
     """Start the given method in a background thread, and add the thread to the TMP_THREADS shutdown hook"""
-    _shutdown_hook = kwargs.pop("_shutdown_hook", True)
-    if not kwargs.get("name"):
-        LOG.debug(
-            "start_thread called without providing a custom name"
-        )  # technically we should add a new level here for *internal* warnings
-    kwargs.setdefault("name", method.__name__)
-    thread = FuncThread(method, *args, **kwargs)
+    if not name:
+        # technically we should add a new level here for *internal* warnings
+        LOG.debug("start_thread called without providing a custom name")
+    name = name or method.__name__
+    thread = FuncThread(method, params=params, quiet=quiet, name=name, on_stop=on_stop)
     thread.start()
     if _shutdown_hook:
         TMP_THREADS.append(thread)
     return thread
 
 
-def start_worker_thread(method, *args, **kwargs):
-    kwargs.setdefault("name", "start_worker_thread")
-    return start_thread(method, *args, _shutdown_hook=False, **kwargs)
+def start_worker_thread(
+    method: "Callable[P, T]", params: Any = None, name: str | None = None
+) -> FuncThread:
+    return start_thread(method, params, _shutdown_hook=False, name=name or "start_worker_thread")
 
 
-def cleanup_threads_and_processes(quiet=True):
+def cleanup_threads_and_processes(quiet: bool = True) -> None:
     from localstack.utils.run import kill_process_tree
 
     for thread in TMP_THREADS:
@@ -153,7 +164,7 @@ def cleanup_threads_and_processes(quiet=True):
     TMP_PROCESSES.clear()
 
 
-def parallelize(func: Callable, arr: list, size: int = None):
+def parallelize(func: Callable, arr: list, size: int = None):  # type: ignore
     if not size:
         size = len(arr)
     if size <= 0:
diff --git a/localstack-core/mypy.ini b/localstack-core/mypy.ini
@@ -1,7 +1,7 @@
 [mypy]
 explicit_package_bases = true
 mypy_path=localstack-core
-files=localstack/aws/api/core.py,localstack/packages,localstack/services/transcribe,localstack/services/kinesis/packages.py
+files=localstack/aws/api/core.py,localstack/utils/threads.py,localstack/utils/strings.py,localstack/packages,localstack/services/transcribe,localstack/services/kinesis/packages.py
 ignore_missing_imports = False
 follow_imports = silent
 ignore_errors = False
diff --git a/tests/unit/test_common.py b/tests/unit/test_common.py
@@ -2,7 +2,6 @@
 import io
 import itertools
 import os
-import threading
 import time
 import zipfile
 from datetime import UTC, date, datetime, timedelta
@@ -363,24 +362,6 @@ def test_format_number(self):
         assert fn(-12.521, decimals=4) == "-12.521"
         assert fn(-1.2234354123e3, decimals=4) == "-1223.4354"
 
-    def test_cleanup_threads_and_processes_calls_shutdown_hooks(self):
-        # TODO: move all run/concurrency related tests into separate class
-
-        started = threading.Event()
-        done = threading.Event()
-
-        def run_method(*args, **kwargs):
-            started.set()
-            func_thread = kwargs["_thread"]
-            # thread waits until it is stopped
-            func_thread._stop_event.wait()
-            done.set()
-
-        common.start_thread(run_method)
-        assert started.wait(timeout=2)
-        common.cleanup_threads_and_processes()
-        assert done.wait(timeout=2)
-
     def test_proxy_map(self):
         old_http_proxy = config.OUTBOUND_HTTP_PROXY
         old_https_proxy = config.OUTBOUND_HTTPS_PROXY
diff --git a/tests/unit/utils/test_threads.py b/tests/unit/utils/test_threads.py
@@ -0,0 +1,62 @@
+import threading
+
+from localstack.utils.threads import (
+    TMP_THREADS,
+    FuncThread,
+    cleanup_threads_and_processes,
+    start_thread,
+    start_worker_thread,
+)
+
+
+class TestThreads:
+    class TestStartThread:
+        def test_start_thread_returns_a_func_thread(self):
+            def examplefunc(*args):
+                pass
+
+            thread = start_thread(examplefunc)
+
+            assert isinstance(thread, FuncThread)
+            assert thread.name.startswith("examplefunc-")
+            assert thread in TMP_THREADS
+
+        def test_start_thread_with_custom_name(self):
+            thread = start_thread(lambda: None, name="somefunc")
+
+            assert thread.name.startswith("somefunc-")
+
+    class TestStartWorkerThread:
+        def test_start_worker_thread_returns_a_func_thread(self):
+            thread = start_worker_thread(lambda: None)
+
+            assert isinstance(thread, FuncThread)
+            assert thread.name.startswith("start_worker_thread-")
+            assert thread not in TMP_THREADS
+
+        def test_start_worker_thread_with_custom_name(self):
+            thread = start_worker_thread(lambda: None, name="somefunc")
+            assert thread.name.startswith("somefunc-")
+
+    def test_cleanup_threads_and_processes_calls_shutdown_hooks(self):
+        started = threading.Event()
+        done = threading.Event()
+
+        # Note: we're extending FuncThread here to make sure we have access to `_stop_event`
+        # Regular users would use `start_thread` instead
+        class ThreadTest(FuncThread):
+            def __init__(self) -> None:
+                super().__init__(self.run_method)
+
+            def run_method(self, *args):
+                started.set()
+                # thread waits until it is stopped
+                self._stop_event.wait()
+                done.set()
+
+        test_thread = ThreadTest()
+        TMP_THREADS.append(test_thread)
+        test_thread.start()
+        assert started.wait(timeout=2)
+        cleanup_threads_and_processes()
+        assert done.wait(timeout=2)
