From 0a2577f6ca939ba01a98885b3d7791c25bae669e Mon Sep 17 00:00:00 2001 From: AlexNg Date: Tue, 19 Dec 2023 14:28:40 +0800 Subject: [PATCH 1/9] + Improved type safety with ParamSpec and TypeVar --- src/thread/_types.py | 10 ++++++++-- src/thread/thread.py | 44 +++++++++++++++++++++++++------------------- 2 files changed, 33 insertions(+), 21 deletions(-) diff --git a/src/thread/_types.py b/src/thread/_types.py index 9204c47..18ffc2a 100644 --- a/src/thread/_types.py +++ b/src/thread/_types.py @@ -5,6 +5,7 @@ """ from typing import Any, Literal, Callable, Union +from typing_extensions import ParamSpec, TypeVar # Descriptive Types @@ -27,5 +28,10 @@ # Function types -HookFunction = Callable[[Data_Out], Union[Any, None]] -TargetFunction = Callable[..., Data_Out] +_Target_P = ParamSpec('_Target_P') +_Target_T = TypeVar('_Target_T') +_Dataset_T = TypeVar('_Dataset_T') + +TargetFunction = Callable[_Target_P, _Target_T] + +HookFunction = Callable[[_Target_T], Union[Any, None]] diff --git a/src/thread/thread.py b/src/thread/thread.py index 92b42a2..0f53402 100644 --- a/src/thread/thread.py +++ b/src/thread/thread.py @@ -19,16 +19,22 @@ class ParallelProcessing: ... from .utils.config import Settings from .utils.algorithm import chunk_split -from ._types import ThreadStatus, Data_In, Data_Out, Overflow_In, TargetFunction, HookFunction +from ._types import ( + ThreadStatus, Data_In, Data_Out, Overflow_In, + TargetFunction, _Target_P, _Target_T, + DatasetFunction, _Dataset_T, + HookFunction +) +from typing_extensions import Generic from typing import ( - Any, List, - Callable, Optional, + Any, List, Unpack, + Callable, Optional, Union, Mapping, Sequence, Tuple ) Threads: set['Thread'] = set() -class Thread(threading.Thread): +class Thread(threading.Thread, Generic[_Target_P, _Target_T]): """ Wraps python's `threading.Thread` class --------------------------------------- @@ -51,7 +57,7 @@ class Thread(threading.Thread): def __init__( self, - target: TargetFunction, + target: TargetFunction[_Target_P, _Target_T], args: Sequence[Data_In] = (), kwargs: Mapping[str, Data_In] = {}, ignore_errors: Sequence[type[Exception]] = (), @@ -100,10 +106,10 @@ def __init__( ) - def _wrap_target(self, target: TargetFunction) -> TargetFunction: + def _wrap_target(self, target: TargetFunction[_Target_P, _Target_T]) -> TargetFunction[_Target_P, Union[_Target_T, None]]: """Wraps the target function""" @wraps(target) - def wrapper(*args: Any, **kwargs: Any) -> Any: + def wrapper(*args: _Target_P.args, **kwargs: _Target_P.kwargs) -> Union[_Target_T, None]: self.status = 'Running' global Threads @@ -173,7 +179,7 @@ def _run_with_trace(self) -> None: @property - def result(self) -> Data_Out: + def result(self) -> _Target_T: """ The return value of the thread @@ -208,7 +214,7 @@ def is_alive(self) -> bool: return super().is_alive() - def add_hook(self, hook: HookFunction) -> None: + def add_hook(self, hook: HookFunction[_Target_T]) -> None: """ Adds a hook to the thread ------------------------- @@ -250,7 +256,7 @@ def join(self, timeout: Optional[float] = None) -> bool: return not self.is_alive() - def get_return_value(self) -> Data_Out: + def get_return_value(self) -> _Target_T: """ Halts the current thread execution until the thread completes @@ -323,7 +329,7 @@ def __init__(self, thread: Thread, progress: float = 0) -> None: self.thread = thread self.progress = progress -class ParallelProcessing: +class ParallelProcessing(Generic[_Target_P, _Target_T, _Dataset_T]): """ Multi-Threaded Parallel Processing --------------------------------------- @@ -335,7 +341,7 @@ class ParallelProcessing: _completed : int status : ThreadStatus - function : Callable[..., List[Data_Out]] + function : TargetFunction[..., List[_Target_T]] dataset : Sequence[Data_In] max_threads : int @@ -344,8 +350,8 @@ class ParallelProcessing: def __init__( self, - function: TargetFunction, - dataset: Sequence[Data_In], + function: DatasetFunction[_Dataset_T, _Target_T], + dataset: Sequence[_Dataset_T], max_threads: int = 8, *overflow_args: Overflow_In, @@ -385,10 +391,10 @@ def __init__( def _wrap_function( self, - function: TargetFunction - ) -> Callable[..., List[Data_Out]]: + function: TargetFunction[[_Dataset_T], _Target_T] + ) -> TargetFunction[..., List[_Target_T]]: @wraps(function) - def wrapper(index: int, data_chunk: Sequence[Data_In], *args: Any, **kwargs: Any) -> List[Data_Out]: + def wrapper(index: int, data_chunk: Sequence[_Dataset_T], *args: _Target_P.args, **kwargs: _Target_P.kwargs) -> List[_Target_T]: computed: List[Data_Out] = [] for i, data_entry in enumerate(data_chunk): v = function(data_entry, *args, **kwargs) @@ -404,7 +410,7 @@ def wrapper(index: int, data_chunk: Sequence[Data_In], *args: Any, **kwargs: Any @property - def results(self) -> Data_Out: + def results(self) -> List[_Dataset_T]: """ The return value of the threads if completed @@ -436,7 +442,7 @@ def is_alive(self) -> bool: return any(entry.thread.is_alive() for entry in self._threads) - def get_return_values(self) -> List[Data_Out]: + def get_return_values(self) -> List[_Dataset_T]: """ Halts the current thread execution until the thread completes From 12d21f32f4a860d84446a60c7135ed01239f3f11 Mon Sep 17 00:00:00 2001 From: AlexNg Date: Tue, 19 Dec 2023 14:29:07 +0800 Subject: [PATCH 2/9] + Improved type safety with ParamSpec and TypeVar --- src/thread/_types.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/thread/_types.py b/src/thread/_types.py index 18ffc2a..2937ef1 100644 --- a/src/thread/_types.py +++ b/src/thread/_types.py @@ -30,8 +30,9 @@ # Function types _Target_P = ParamSpec('_Target_P') _Target_T = TypeVar('_Target_T') -_Dataset_T = TypeVar('_Dataset_T') - TargetFunction = Callable[_Target_P, _Target_T] HookFunction = Callable[[_Target_T], Union[Any, None]] + +_Dataset_T = TypeVar('_Dataset_T') +DatasetFunction = Callable[[_Dataset_T], _Target_T] From b76cedc90eac359b820b17de38c77daca9dec418 Mon Sep 17 00:00:00 2001 From: AlexNg Date: Wed, 20 Dec 2023 09:47:06 +0800 Subject: [PATCH 3/9] - Removed unused imports --- src/thread/thread.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/thread/thread.py b/src/thread/thread.py index 0f53402..202d2d7 100644 --- a/src/thread/thread.py +++ b/src/thread/thread.py @@ -27,7 +27,7 @@ class ParallelProcessing: ... ) from typing_extensions import Generic from typing import ( - Any, List, Unpack, + List, Callable, Optional, Union, Mapping, Sequence, Tuple ) From 9b210919b1de4ce59aaa27e99c2e359b90e4a8d0 Mon Sep 17 00:00:00 2001 From: AlexNg Date: Fri, 22 Dec 2023 00:36:06 +0800 Subject: [PATCH 4/9] Fix failing tests for python3.9 --- src/thread/thread.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/src/thread/thread.py b/src/thread/thread.py index 202d2d7..607aeeb 100644 --- a/src/thread/thread.py +++ b/src/thread/thread.py @@ -25,7 +25,7 @@ class ParallelProcessing: ... DatasetFunction, _Dataset_T, HookFunction ) -from typing_extensions import Generic +from typing_extensions import Generic, Concatenate, ParamSpec from typing import ( List, Callable, Optional, Union, @@ -321,6 +321,7 @@ def start(self) -> None: +_P = ParamSpec('_P') class _ThreadWorker: progress: float thread: Thread @@ -341,7 +342,7 @@ class ParallelProcessing(Generic[_Target_P, _Target_T, _Dataset_T]): _completed : int status : ThreadStatus - function : TargetFunction[..., List[_Target_T]] + function : TargetFunction dataset : Sequence[Data_In] max_threads : int @@ -391,8 +392,8 @@ def __init__( def _wrap_function( self, - function: TargetFunction[[_Dataset_T], _Target_T] - ) -> TargetFunction[..., List[_Target_T]]: + function: TargetFunction + ) -> TargetFunction: @wraps(function) def wrapper(index: int, data_chunk: Sequence[_Dataset_T], *args: _Target_P.args, **kwargs: _Target_P.kwargs) -> List[_Target_T]: computed: List[Data_Out] = [] @@ -512,6 +513,8 @@ def start(self) -> None: name_format = self.overflow_kwargs.get('name') and self.overflow_kwargs['name'] + '%s' self.overflow_kwargs = { i: v for i,v in self.overflow_kwargs.items() if i != 'name' and i != 'args' } + print(parsed_args, self.overflow_args) + for i, data_chunk in enumerate(chunk_split(self.dataset, max_threads)): chunk_thread = Thread( target = self.function, From 226aa1defd53f1dfbce673aeb4b771b284b3e73b Mon Sep 17 00:00:00 2001 From: AlexNg Date: Sun, 24 Dec 2023 10:54:08 +0800 Subject: [PATCH 5/9] - Removed unused functions --- tests/test_decorator.py | 9 --------- 1 file changed, 9 deletions(-) diff --git a/tests/test_decorator.py b/tests/test_decorator.py index 53117d2..0af6c09 100644 --- a/tests/test_decorator.py +++ b/tests/test_decorator.py @@ -8,15 +8,6 @@ def _dummy_target_raiseToPower(x: float, power: float, delay: float = 0): time.sleep(delay) return x**power -def _dummy_raiseException(x: Exception, delay: float = 0): - time.sleep(delay) - raise x - -def _dummy_iterative(itemCount: int, pTime: float = 0.1, delay: float = 0): - time.sleep(delay) - for i in range(itemCount): - time.sleep(pTime) - From 3281e6b11f813d0c41f4e3ee26d219dfd5cf20ae Mon Sep 17 00:00:00 2001 From: AlexNg Date: Sun, 24 Dec 2023 10:54:50 +0800 Subject: [PATCH 6/9] + Made more type-safe, Closes #32 --- src/thread/decorators.py | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/src/thread/decorators.py b/src/thread/decorators.py index 4755d3c..2e48109 100644 --- a/src/thread/decorators.py +++ b/src/thread/decorators.py @@ -15,14 +15,14 @@ T = TypeVar('T') P = ParamSpec('P') TargetFunction = Callable[P, Data_Out] -NoParamReturn = Callable[P, Thread] -WithParamReturn = Callable[[TargetFunction], NoParamReturn] +NoParamReturn = Callable[P, Thread[P, T]] +WithParamReturn = Callable[[TargetFunction[P]], NoParamReturn[P, T]] FullParamReturn = Callable[P, Thread] -WrappedWithParamReturn = Callable[[TargetFunction], WithParamReturn] +WrappedWithParamReturn = Callable[[TargetFunction[P]], WithParamReturn[P, T]] @overload -def threaded(__function: TargetFunction) -> NoParamReturn: ... +def threaded(__function: TargetFunction[P]) -> NoParamReturn[P, T]: ... @overload def threaded( @@ -32,7 +32,7 @@ def threaded( ignore_errors: Sequence[type[Exception]] = (), suppress_errors: bool = False, **overflow_kwargs: Overflow_In -) -> WithParamReturn: ... +) -> WithParamReturn[P, T]: ... @overload def threaded( @@ -43,18 +43,18 @@ def threaded( ignore_errors: Sequence[type[Exception]] = (), suppress_errors: bool = False, **overflow_kwargs: Overflow_In -) -> FullParamReturn: ... +) -> FullParamReturn[P]: ... def threaded( - __function: Optional[TargetFunction] = None, + __function: Optional[TargetFunction[P]] = None, *, args: Sequence[Data_In] = (), kwargs: Mapping[str, Data_In] = {}, ignore_errors: Sequence[type[Exception]] = (), suppress_errors: bool = False, **overflow_kwargs: Overflow_In -) -> Union[NoParamReturn, WithParamReturn, FullParamReturn]: +) -> Union[NoParamReturn[P, T], WithParamReturn[P, T], FullParamReturn[P]]: """ Decorate a function to run it in a thread @@ -96,7 +96,7 @@ def threaded( """ if not callable(__function): - def wrapper(func: TargetFunction) -> FullParamReturn: + def wrapper(func: TargetFunction[P]) -> FullParamReturn[P]: return threaded( func, args = args, From a6b16d196a00448a67c6b1c20c5378c94e263d0d Mon Sep 17 00:00:00 2001 From: AlexNg Date: Sun, 24 Dec 2023 10:59:53 +0800 Subject: [PATCH 7/9] - Remove unused imports --- src/thread/thread.py | 2 +- tests/test_decorator.py | 3 +-- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/src/thread/thread.py b/src/thread/thread.py index 607aeeb..7e14c19 100644 --- a/src/thread/thread.py +++ b/src/thread/thread.py @@ -25,7 +25,7 @@ class ParallelProcessing: ... DatasetFunction, _Dataset_T, HookFunction ) -from typing_extensions import Generic, Concatenate, ParamSpec +from typing_extensions import Generic, ParamSpec from typing import ( List, Callable, Optional, Union, diff --git a/tests/test_decorator.py b/tests/test_decorator.py index 0af6c09..c83541c 100644 --- a/tests/test_decorator.py +++ b/tests/test_decorator.py @@ -1,6 +1,5 @@ import time -import pytest -from src.thread import threaded, exceptions +from src.thread import threaded # >>>>>>>>>> Dummy Functions <<<<<<<<<< # From df3021b1ac81aa0965aa5e42d2d342c64ee890d3 Mon Sep 17 00:00:00 2001 From: AlexNg Date: Sun, 24 Dec 2023 11:19:43 +0800 Subject: [PATCH 8/9] Made returned_value private --- src/thread/thread.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/thread/thread.py b/src/thread/thread.py index 7e14c19..861e958 100644 --- a/src/thread/thread.py +++ b/src/thread/thread.py @@ -44,7 +44,7 @@ class Thread(threading.Thread, Generic[_Target_P, _Target_T]): status : ThreadStatus hooks : List[HookFunction] - returned_value: Data_Out + _returned_value: Data_Out errors : List[Exception] ignore_errors : Sequence[type[Exception]] @@ -86,7 +86,7 @@ def __init__( :param **: These are arguments parsed to `thread.Thread` """ _target = self._wrap_target(target) - self.returned_value = None + self._returned_value = None self.status = 'Idle' self.hooks = [] @@ -116,7 +116,7 @@ def wrapper(*args: _Target_P.args, **kwargs: _Target_P.kwargs) -> Union[_Target_ Threads.add(self) try: - self.returned_value = target(*args, **kwargs) + self._returned_value = target(*args, **kwargs) except Exception as e: if not any(isinstance(e, ignore) for ignore in self.ignore_errors): self.status = 'Errored' @@ -135,7 +135,7 @@ def _invoke_hooks(self) -> None: errors: List[Tuple[Exception, str]] = [] for hook in self.hooks: try: - hook(self.returned_value) + hook(self._returned_value) except Exception as e: if not any(isinstance(e, ignore) for ignore in self.ignore_errors): errors.append(( @@ -196,7 +196,7 @@ def result(self) -> _Target_T: self._handle_exceptions() if self.status in ['Invoking hooks', 'Completed']: - return self.returned_value + return self._returned_value else: raise exceptions.ThreadStillRunningError() From adb2f134a6461f22ae52e59d990d427ea1a49373 Mon Sep 17 00:00:00 2001 From: AlexNg Date: Sun, 24 Dec 2023 11:24:45 +0800 Subject: [PATCH 9/9] Made more type safe --- src/thread/decorators.py | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/src/thread/decorators.py b/src/thread/decorators.py index 2e48109..f610e42 100644 --- a/src/thread/decorators.py +++ b/src/thread/decorators.py @@ -14,15 +14,15 @@ T = TypeVar('T') P = ParamSpec('P') -TargetFunction = Callable[P, Data_Out] +TargetFunction = Callable[P, T] NoParamReturn = Callable[P, Thread[P, T]] -WithParamReturn = Callable[[TargetFunction[P]], NoParamReturn[P, T]] -FullParamReturn = Callable[P, Thread] -WrappedWithParamReturn = Callable[[TargetFunction[P]], WithParamReturn[P, T]] +WithParamReturn = Callable[[TargetFunction[P, T]], NoParamReturn[P, T]] +FullParamReturn = Callable[P, Thread[P, T]] +WrappedWithParamReturn = Callable[[TargetFunction[P, T]], WithParamReturn[P, T]] @overload -def threaded(__function: TargetFunction[P]) -> NoParamReturn[P, T]: ... +def threaded(__function: TargetFunction[P, T]) -> NoParamReturn[P, T]: ... @overload def threaded( @@ -36,25 +36,25 @@ def threaded( @overload def threaded( - __function: Callable[P, Data_Out], + __function: TargetFunction[P, T], *, args: Sequence[Data_In] = (), kwargs: Mapping[str, Data_In] = {}, ignore_errors: Sequence[type[Exception]] = (), suppress_errors: bool = False, **overflow_kwargs: Overflow_In -) -> FullParamReturn[P]: ... +) -> FullParamReturn[P, T]: ... def threaded( - __function: Optional[TargetFunction[P]] = None, + __function: Optional[TargetFunction[P, T]] = None, *, args: Sequence[Data_In] = (), kwargs: Mapping[str, Data_In] = {}, ignore_errors: Sequence[type[Exception]] = (), suppress_errors: bool = False, **overflow_kwargs: Overflow_In -) -> Union[NoParamReturn[P, T], WithParamReturn[P, T], FullParamReturn[P]]: +) -> Union[NoParamReturn[P, T], WithParamReturn[P, T], FullParamReturn[P, T]]: """ Decorate a function to run it in a thread @@ -96,7 +96,7 @@ def threaded( """ if not callable(__function): - def wrapper(func: TargetFunction[P]) -> FullParamReturn[P]: + def wrapper(func: TargetFunction[P, T]) -> FullParamReturn[P, T]: return threaded( func, args = args, @@ -115,7 +115,7 @@ def wrapper(func: TargetFunction[P]) -> FullParamReturn[P]: kwargs = dict(kwargs) @wraps(__function) - def wrapped(*parsed_args: P.args, **parsed_kwargs: P.kwargs) -> Thread: + def wrapped(*parsed_args: P.args, **parsed_kwargs: P.kwargs) -> Thread[P, T]: kwargs.update(parsed_kwargs) processed_args = ( *args, *parsed_args )