perf($mypy): type var for return type

python_boilerplate/common/asynchronization.py

@@ -1,7 +1,7 @@
 import functools
 import inspect
 from concurrent.futures import Future
-from typing import Any, Callable
+from typing import Any, Callable, TypeVar
 
 from loguru import logger
 
@@ -10,8 +10,10 @@
     executor,
 )
 
+R = TypeVar("R")
 
-def async_function(func: Callable[..., Any]) -> Callable[..., Future[Any]]:
+
+def async_function(func: Callable[..., R]) -> Callable[..., Future[R]]:
     """
     An easy way to implement multi-tread feature with thread pool. The decorator to run function in thread pool.
     The return value of decorated function will be `concurrent.futures._base.Future`.
@@ -34,7 +36,7 @@ def async_function(func: Callable[..., Any]) -> Callable[..., Future[Any]]:
     """
 
     @functools.wraps(func)
-    def wrapped(*arg: Any, **kwarg: Any) -> Future[Any]:
+    def wrapped(*arg: Any, **kwarg: Any) -> Future[R]:
         module = inspect.getmodule(func)
         if arg and not kwarg:
             submitted_future = executor.submit(func, *arg)

python_boilerplate/demo/multithread_and_thread_pool_usage.py

@@ -1,5 +1,5 @@
 import random
-from concurrent.futures import wait
+from concurrent.futures import Future, wait
 
 import pandas as pd
 from faker import Faker
@@ -44,7 +44,7 @@ def generate_data_frame(rows: int) -> DataFrame:
 def async_generate_data_frame() -> DataFrame:
     total_rows = sum(ROW_ARRAY)
     logger.info(f"Going to generate {total_rows} rows of data in asynchronous way")
-    futures = [generate_data_frame(row) for row in ROW_ARRAY]
+    futures: list[Future[DataFrame]] = [generate_data_frame(row) for row in ROW_ARRAY]
     logger.info(
         f"Submitted {len(futures)} tasks to ThreadPoolExecutor, waiting for them to complete..."
     )