Improve process future tasks command #18
Conversation
christophbuermann
force-pushed
the
improve_process_future_tasks_command
branch
from
eaeb63f to
51a2bd6
Compare
christophbuermann
force-pushed
the
improve_process_future_tasks_command
branch
from
51a2bd6 to
376878a
Compare
|
|
| def _endless_task_iterator(self): | ||
| while self._running: | ||
| tasks = self._get_open_tasks() | ||
| yield from tasks | ||
| if not tasks: | ||
| time.sleep(self.wait_for_tasks_duration_seconds) |
There was a problem hiding this comment.
This is the new basis for the worker-behavior.
| def _handle_task(self, task): | ||
| task.status = FutureTask.FUTURE_TASK_STATUS_IN_PROGRESS | ||
| task.save() | ||
| self.current_task_pk = task.pk |
There was a problem hiding this comment.
Instead of handling batches of tasks, we now handle one task in one "tick".
| parser.add_argument( | ||
| "--one-time-run", | ||
| action="store_true", | ||
| help="Process tasks that are open at the time of running the command and exit.", | ||
| ) |
There was a problem hiding this comment.
Fixed the one-time-run argument. The old one was broken.
| parser.add_argument( | ||
| "--wait-for-tasks-duration-seconds", | ||
| type=float, | ||
| default=1.0, | ||
| help="If there are no open tasks the command waits this amount of time until it checks for open tasks again.", | ||
| ) |
There was a problem hiding this comment.
New argument for the waiting time.
|
|
||
|
|
||
| def _wait_for_task_status(task, status, tick_seconds=0.1, timeout_seconds=3): | ||
| def _wait_for_task_status(task, status, tick_seconds: float = 0.15, timeout_seconds: float = 3.0): |
There was a problem hiding this comment.
Locking shenanigans with SQLite caused me to use 0.15 seconds as default waiting time because it doesn't align so perfectly with the default 1.0 waiting time of the worker. Postgres would be nice.
| @pytest.mark.django_db(transaction=True) | ||
| class TestWorker: |
There was a problem hiding this comment.
I did some pytest-friendly refactoring of the tests.
| class ProcessFutureTasksWorker: | ||
| def __init__(self, *args: str): |
There was a problem hiding this comment.
Helpful worker with context manager interface for usage in the tests.
1 participant
process_future_taskscommand.