fork/django-q2
1
0
Fork 0
mirror of https://github.com/django-q2/django-q2.git synced 2026-09-15 13:37:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

Show function name in log when running task (#3)

Browse Source 
* refactor: show function name as it runs

* feat: log func name when processed

Co-authored-by: Noortheen Raja <jnoortheen@gmail.com>
This commit is contained in:
Stan Triepels
2022-09-17 01:48:07 +02:00
committed by GitHub
parent 0b9703f56a
commit c4ebfc66b9
1 changed files with 9 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
django_q/cluster.py
View File

@@ -389,12 +389,13 @@ def monitor(result_queue: Queue, broker: Broker = None):
# signal execution done # signal execution done
post_execute.send(sender="django_q", task=task) post_execute.send(sender="django_q", task=task)
# log the result # log the result
info_name = f"{task['name']} ({task['func']})"
if task["success"]: if task["success"]:
# log success # log success
logger.info(_(f"Processed [{task['name']}]")) logger.info(_(f"Processed [{info_name}]"))
else: else:
# log failure # log failure
logger.error(_(f"Failed [{task['name']}] - {task['result']}")) logger.error(_(f"Failed [{info_name}] - {task['result']}"))
logger.info(_(f"{name} stopped monitoring results")) logger.info(_(f"{name} stopped monitoring results"))
@@ -408,8 +409,8 @@ def worker(
:type result_queue: multiprocessing.Queue :type result_queue: multiprocessing.Queue
:type timer: multiprocessing.Value :type timer: multiprocessing.Value
""" """
name = current_process().name proc_name = current_process().name
logger.info(_(f"{name} ready for work at {current_process().pid}")) logger.info(_(f"{proc_name} ready for work at {current_process().pid}"))
task_count = 0 task_count = 0
if timeout is None: if timeout is None:
timeout = -1 timeout = -1
@@ -419,7 +420,9 @@ def worker(
timer.value = -1 # Idle timer.value = -1 # Idle
task_count += 1 task_count += 1
# Get the function from the task # Get the function from the task
logger.info(_(f'{name} processing [{task["name"]}]')) func = task["func"]
func_name = func.__name__ if hasattr(func, "__name__") else str(func)
logger.info(_(f'{proc_name} processing [{task["name"]}({func_name})]'))
f = task["func"] f = task["func"]
# if it's not an instance try to get it from the string # if it's not an instance try to get it from the string
if not callable(task["func"]): if not callable(task["func"]):
@@ -450,7 +453,7 @@ def worker(
if task_count == Conf.RECYCLE or rss_check(): if task_count == Conf.RECYCLE or rss_check():
timer.value = -2 # Recycled timer.value = -2 # Recycled
break break
logger.info(_(f"{name} stopped doing work")) logger.info(_(f"{proc_name} stopped doing work"))
def save_task(task, broker: Broker): def save_task(task, broker: Broker):