hoelee/tubesync
1
0
Fork 0
mirror of https://github.com/meeb/tubesync.git synced 2025-06-23 05:26:37 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

WIP: sketch out the register_* functions on the wrapper class

Browse Source
This commit is contained in:
tcely 2025-06-02 07:49:41 -04:00 committed by GitHub
parent a95b201f0c
commit 237367dff3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 14 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
tubesync/common/huey.py
View File

@ -10,9 +10,20 @@ from huey.storage import SqliteStorage
class CompatibleTaskWrapper(TaskWrapper): class CompatibleTaskWrapper(TaskWrapper):
registry = defaultdict(dict) registry = defaultdict(dict)
def backoff(self, attempt, /): @staticmethod
def backoff(attempt, /):
return (5+(attempt**4)) return (5+(attempt**4))
@classmethod
def register_function(cls, fn, /, *args, **kwargs):
def validate_datetime(timestamp):
return False
return partialmethod(validate_datetime)
@classmethod
def register_instance(cls, fn, wrapper, /, *args, **kwargs):
pass
def backoff_list(self, retries, /): def backoff_list(self, retries, /):
if not self._backoff_list: if not self._backoff_list:
self._backoff_list = [ self.backoff(a) for a in range(retries, -1, -1) ] self._backoff_list = [ self.backoff(a) for a in range(retries, -1, -1) ]
@ -188,12 +199,12 @@ def background(name=None, schedule=None, queue=None, remove_existing_tasks=False
def _decorator(fn): def _decorator(fn):
return db_task(**task_args)(fn) return db_task(**task_args)(fn)
def _periodic_decorator(fn): def _periodic_decorator(fn):
when_to_run_function = TaskWrapper.register_function(fn, task_args) when_to_run_function = TaskWrapper.register_function(fn, **task_args)
return db_periodic_task(when_to_run_function, **task_args)(fn) return db_periodic_task(when_to_run_function, **task_args)(fn)
if fn: if fn:
if repeats: if repeats:
wrapper = _periodic_decorator(fn) wrapper = _periodic_decorator(fn)
TaskWrapper.register_instance(fn, task_args, wrapper) TaskWrapper.register_instance(fn, wrapper, **task_args)
else: else:
wrapper = _decorator(fn) wrapper = _decorator(fn)
wrapper.now = wrapper.call_local wrapper.now = wrapper.call_local