From 05ed2745b06d77f3fb0402dcbc02c7032c452291 Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Wed, 5 Nov 2025 13:05:57 +0800 Subject: [PATCH] =?UTF-8?q?modify:=20=E4=BF=AE=E6=94=B9=E9=87=8D=E8=AF=95?= =?UTF-8?q?=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .python-version | 2 +- pyproject.toml | 28 ++++++++++++++-------- src/queue_sqlite/cycle/async_task_cycle.py | 13 ++++++---- src/queue_sqlite/cycle/task_cycle.py | 14 +++++++---- src/queue_sqlite/mounter/task_mounter.py | 18 +++++++++++++- src/queue_sqlite_core/pyproject.toml | 2 +- 6 files changed, 54 insertions(+), 23 deletions(-) diff --git a/.python-version b/.python-version index 24ee5b1..2c07333 100644 --- a/.python-version +++ b/.python-version @@ -1 +1 @@ -3.13 +3.11 diff --git a/pyproject.toml b/pyproject.toml index c7ce9d0..e1d825e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,7 +13,7 @@ classifiers = [ "Operating System :: POSIX :: Linux", ] dependencies = [ - "queue_sqlite_core>=0.2.0", + "queue-sqlite-core>=0.2.0", ] description = "A high-performance, SQLite-based distributed task queue system with Rust-backed core operations. Supports task mounting, message listening, priority handling, retry mechanisms, and automatic cleanup of expired messages. Ideal for building reliable, scalable background task processing systems." name = "queue-sqlite" @@ -22,13 +22,13 @@ requires-python = ">=3.11" version = "0.2.1" [project.optional-dependencies] -pyside6 = ["qtpy", "PySide6"] -pyqt6 = ["qtpy", "PyQt6"] pyqt5 = ["qtpy", "PyQt5"] +pyqt6 = ["qtpy", "PyQt6"] +pyside6 = ["qtpy", "PySide6"] -[[tool.uv.index]] -default = true -url = "https://pypi.tuna.tsinghua.edu.cn/simple" +# [[tool.uv.index]] +# default = true +# url = "https://pypi.tuna.tsinghua.edu.cn/simple" [build-system] build-backend = "hatchling.build" @@ -36,11 +36,19 @@ requires = ["hatchling"] [dependency-groups] dev = [ - "maturin>=1.9.4", - "psutil>=7.0.0", - "pytest>=7.4.4", - "twine>=4.0.2", + "maturin>=1.9.4", + "psutil>=7.0.0", + "pytest>=7.4.4", + "twine>=4.0.2", ] [tool.hatch.build.targets.wheel] packages = ["src/queue_sqlite"] + +[tool.uv.workspace] +members = [ + "src/queue_sqlite_core", +] + +[tool.uv.sources] +queue-sqlite-core = { workspace = true } diff --git a/src/queue_sqlite/cycle/async_task_cycle.py b/src/queue_sqlite/cycle/async_task_cycle.py index 48ae462..597f705 100644 --- a/src/queue_sqlite/cycle/async_task_cycle.py +++ b/src/queue_sqlite/cycle/async_task_cycle.py @@ -11,12 +11,13 @@ from ..model import MessageItem from ..constant import MessageStatus +from ..mounter.task_mounter import TaskMeta import json import asyncio import functools -def retry_async(max_retries=3, delay=1): +def retry_async(max_retries=3): """ 异步重试装饰器 @@ -30,8 +31,10 @@ def retry_async(max_retries=3, delay=1): async def wrapper(self, *args, **kwargs): last_exception = None # 使用message_item中的retry_count作为最大重试次数 - retries = getattr(self.message_item, "retry_count", max_retries) - + task_meta: TaskMeta = self.callback.meta + retries = task_meta.max_retries + # retries = getattr(self.message_item, "retry_count", max_retries) + delay_time = task_meta.delay # 至少尝试一次(retries+1),最多尝试max_retries+1次 actual_retries = min(retries, max_retries) if max_retries > 0 else retries @@ -41,7 +44,7 @@ def retry_async(max_retries=3, delay=1): except Exception as e: last_exception = e if attempt < actual_retries: - await asyncio.sleep(delay) + await asyncio.sleep(delay_time) else: break if last_exception: @@ -62,7 +65,7 @@ class AsyncTaskCycle: self.task_status = None self.task_error = None - @retry_async(max_retries=3, delay=1) + @retry_async(max_retries=3) async def run(self): try: task_result = await self.callback(self.message_item) # type: ignore diff --git a/src/queue_sqlite/cycle/task_cycle.py b/src/queue_sqlite/cycle/task_cycle.py index 44ac917..5dae02b 100644 --- a/src/queue_sqlite/cycle/task_cycle.py +++ b/src/queue_sqlite/cycle/task_cycle.py @@ -12,13 +12,14 @@ from ..model import MessageItem from typing import Callable, Optional from ..constant import MessageStatus +from ..mounter.task_mounter import TaskMeta import json import functools import time -def retry_sync(max_retries=3, delay=1): +def retry_sync(max_retries=3): """ 同步重试装饰器 @@ -32,8 +33,10 @@ def retry_sync(max_retries=3, delay=1): def wrapper(self, *args, **kwargs): last_exception = None # 使用message_item中的retry_count作为最大重试次数 - retries = getattr(self.message_item, "retry_count", max_retries) - + task_meta: TaskMeta = self.callback.meta + retries = task_meta.max_retries + # retries = getattr(self.message_item, "retry_count", max_retries) + delay_time = task_meta.delay # 至少尝试一次(retries+1),最多尝试max_retries+1次 actual_retries = min(retries, max_retries) if max_retries > 0 else retries @@ -42,8 +45,9 @@ def retry_sync(max_retries=3, delay=1): return func(self, *args, **kwargs) except Exception as e: last_exception = e + self.message_item.retry_count = attempt if attempt < actual_retries: - time.sleep(delay) + time.sleep(delay_time) else: break @@ -66,7 +70,7 @@ class TaskCycle: self.task_status = None self.task_error = None - @retry_sync(max_retries=3, delay=1) + @retry_sync(max_retries=3) def run(self): try: task_result = self.callback(self.message_item) # type: ignore diff --git a/src/queue_sqlite/mounter/task_mounter.py b/src/queue_sqlite/mounter/task_mounter.py index 9ab8b0e..27677e5 100644 --- a/src/queue_sqlite/mounter/task_mounter.py +++ b/src/queue_sqlite/mounter/task_mounter.py @@ -10,6 +10,15 @@ from typing import Callable, List +from dataclasses import dataclass, field + + +@dataclass +class TaskMeta: + name: str = field(default="", metadata={"help": "任务名称"}) + description: str = field(default="", metadata={"help": "任务描述"}) + max_retries: int = field(default=3, metadata={"help": "任务最大重试次数"}) + delay: float = field(default=1, metadata={"help": "任务延迟执行时间"}) class TaskMounter: @@ -23,7 +32,14 @@ class TaskMounter: def decorator(function: Callable): # 使用自定义名称或函数原名 - function.meta = meta # type: ignore + name = meta.get("name", function.__name__) + description = meta.get("description", "") + max_retries = meta.get("max_retries", 3) + delay = meta.get("delay", 1) + task_meta = TaskMeta( + name=name, description=description, max_retries=max_retries, delay=delay + ) + function.meta = task_meta # type: ignore setattr(TaskMounter, function.__name__, function) return function diff --git a/src/queue_sqlite_core/pyproject.toml b/src/queue_sqlite_core/pyproject.toml index abe82d3..d2457be 100644 --- a/src/queue_sqlite_core/pyproject.toml +++ b/src/queue_sqlite_core/pyproject.toml @@ -19,7 +19,7 @@ features = ["pyo3/extension-module"] [[tool.uv.index]] default = true -url = "http://124.71.68.6:3141/root/pypi" +url = "https://pypi.tuna.tsinghua.edu.cn/simple" [dependency-groups] dev = [