modify: 修改重试逻辑

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

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

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
 

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 = [