$ pip install huey
python
from huey import RedisHuey
huey = RedisHuey(url='redis://localhost:6379/0')
python
from huey import RedisHuey
huey = RedisHuey(url='redis://localhost:6379/0')
@huey.task()
def my_task(param1, param2):
with open('log.txt', 'a') as f:
f.write(f"Task executed with params: {param1}, {param2}
")
python
from huey import Plugin
class ExecutionTimePlugin(Plugin):
def pre_execute(self, task):
task._start_time = time.time()
def post_execute(self, task, success, exc=None):
execution_time = time.time() - task._start_time
with open('execution_log.txt', 'a') as f:
f.write(f"Task {task.id} executed in {execution_time} seconds
")
huey.use([ExecutionTimePlugin()])
python
from huey import RedisHuey
from huey.exceptions import RetryTask
huey = RedisHuey(url='redis://localhost:6379/0')
@huey.task(retries=3, retry_delay=10, timeout=30)
def my_task(param):
try:
pass
except Exception as e:
raise RetryTask(e)
