diff --git a/py_src/taskito/canvas.py b/py_src/taskito/canvas.py index dc0e7877..6db36e27 100644 --- a/py_src/taskito/canvas.py +++ b/py_src/taskito/canvas.py @@ -79,9 +79,7 @@ class group: concurrently. Members are dispatched in waves. """ - def __init__( - self, *signatures: Signature, max_concurrency: int | None = None - ): + def __init__(self, *signatures: Signature, max_concurrency: int | None = None): if len(signatures) < 1: raise ValueError("group requires at least one signature") self.signatures = list(signatures) @@ -154,9 +152,7 @@ def apply(self, queue: Queue | None = None) -> JobResult: ) -def chunks( - task: TaskWrapper, items: list[Any], chunk_size: int -) -> group: +def chunks(task: TaskWrapper, items: list[Any], chunk_size: int) -> group: """Split items into chunks and create a group of tasks processing each chunk. Args: @@ -179,9 +175,7 @@ def chunks( return group(*sigs) -def starmap( - task: TaskWrapper, args_list: list[tuple[Any, ...]] -) -> group: +def starmap(task: TaskWrapper, args_list: list[tuple[Any, ...]]) -> group: """Create a group with one task per args tuple. Args: diff --git a/py_src/taskito/context.py b/py_src/taskito/context.py index cc731b8c..d5754031 100644 --- a/py_src/taskito/context.py +++ b/py_src/taskito/context.py @@ -75,9 +75,7 @@ def log( if _queue_ref is None: raise RuntimeError("Queue reference not set. Cannot write log.") extra_str = json.dumps(extra) if extra else None - _queue_ref._inner.write_task_log( - ctx.job_id, ctx.task_name, level, message, extra_str - ) + _queue_ref._inner.write_task_log(ctx.job_id, ctx.task_name, level, message, extra_str) def check_cancelled(self) -> None: """Check if cancellation has been requested for this job. diff --git a/py_src/taskito/dashboard.py b/py_src/taskito/dashboard.py index edfb2e1b..cda8fee2 100644 --- a/py_src/taskito/dashboard.py +++ b/py_src/taskito/dashboard.py @@ -25,7 +25,9 @@ def _load_spa_html() -> str: """Load the dashboard SPA HTML from the bundled template file.""" - return resources.files("taskito").joinpath("templates/dashboard.html").read_text(encoding="utf-8") + return ( + resources.files("taskito").joinpath("templates/dashboard.html").read_text(encoding="utf-8") + ) _SPA_HTML: str | None = None @@ -106,7 +108,9 @@ def do_GET(self) -> None: level = qs.get("level", [None])[0] since = int(qs.get("since", ["3600"])[0]) limit = int(qs.get("limit", ["100"])[0]) - self._json_response(queue.query_logs(task_name=task, level=level, since=since, limit=limit)) + self._json_response( + queue.query_logs(task_name=task, level=level, since=since, limit=limit) + ) elif path == "/api/circuit-breakers": self._json_response(queue.circuit_breakers()) elif path == "/api/workers": diff --git a/py_src/taskito/mixins.py b/py_src/taskito/mixins.py index 44ab0ef9..e8b8203b 100644 --- a/py_src/taskito/mixins.py +++ b/py_src/taskito/mixins.py @@ -19,9 +19,7 @@ class _AsyncMixin: async def _run_sync(self, fn: Any, *args: Any, **kwargs: Any) -> Any: loop = asyncio.get_event_loop() if kwargs: - return await loop.run_in_executor( - self._executor, lambda: fn(*args, **kwargs) - ) + return await loop.run_in_executor(self._executor, lambda: fn(*args, **kwargs)) return await loop.run_in_executor(self._executor, lambda: fn(*args)) diff --git a/py_src/taskito/result.py b/py_src/taskito/result.py index 26cfc450..41f378ce 100644 --- a/py_src/taskito/result.py +++ b/py_src/taskito/result.py @@ -90,9 +90,7 @@ def _poll_once(self) -> tuple[str, Any]: return status, cloudpickle.loads(result_bytes) if status in ("failed", "dead"): - raise RuntimeError( - f"Job {self.id} {status}: {self._py_job.error or 'unknown error'}" - ) + raise RuntimeError(f"Job {self.id} {status}: {self._py_job.error or 'unknown error'}") return status, None diff --git a/py_src/taskito/testing.py b/py_src/taskito/testing.py index 9b7cc6ab..915fac8a 100644 --- a/py_src/taskito/testing.py +++ b/py_src/taskito/testing.py @@ -181,9 +181,7 @@ def progress(self) -> int | None: def result(self, timeout: float = 30.0, **kwargs: Any) -> Any: if self._tr.error: - raise RuntimeError( - f"Job {self._tr.job_id} failed: {self._tr.error}" - ) + raise RuntimeError(f"Job {self._tr.job_id} failed: {self._tr.error}") return self._tr.return_value async def aresult(self, timeout: float = 30.0, **kwargs: Any) -> Any: diff --git a/pyproject.toml b/pyproject.toml index 974ef5bd..4b63f4c2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,7 +23,7 @@ module-name = "taskito._taskito" features = ["pyo3/extension-module"] [project.scripts] -taskito = "taskito.cli:main" +taskito = "taskito.cli:main" [tool.pytest.ini_options] asyncio_mode = "auto" diff --git a/uv.lock b/uv.lock index f559a30a..082289eb 100644 --- a/uv.lock +++ b/uv.lock @@ -791,7 +791,7 @@ wheels = [ [[package]] name = "taskito" -version = "0.1.1" +version = "0.2.0" source = { editable = "." } dependencies = [ { name = "cloudpickle" },