From 3d058d6603647e4c69bbdcce95e021f0bb7394ea Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micha=C5=82=20Praszmo?= Date: Mon, 25 Mar 2024 13:58:53 +0100 Subject: [PATCH] Simplify types --- karton/core/karton.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/karton/core/karton.py b/karton/core/karton.py index 1a87ee0..3001428 100644 --- a/karton/core/karton.py +++ b/karton/core/karton.py @@ -17,6 +17,8 @@ from .task import Task, TaskState from .utils import timeout +KartonTaskException = Union[Exception, BaseException] + class Producer(KartonBase): """ @@ -132,7 +134,7 @@ def __init__( self._post_hooks: List[ Tuple[ Optional[str], - Callable[[Task, Optional[Union[Exception, BaseException]]], None], + Callable[[Task, Optional[KartonTaskException]], None], ] ] = [] @@ -264,7 +266,7 @@ def add_pre_hook( def add_post_hook( self, - callback: Callable[[Task, Optional[Exception]], None], + callback: Callable[[Task, Optional[KartonTaskException]], None], name: Optional[str] = None, ) -> None: """ @@ -293,9 +295,7 @@ def _run_pre_hooks(self) -> None: else: self.log.exception("Pre-hook failed") - def _run_post_hooks( - self, exception: Optional[Union[Exception, BaseException]] - ) -> None: + def _run_post_hooks(self, exception: Optional[KartonTaskException]) -> None: """ Run registered postprocessing hooks