Skip to content

Add the ability to run parallel tasks #200

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
rroblf01 wants to merge 12 commits into
base: master
Choose a base branch
from
Open

Add the ability to run parallel tasks #200

Changes from 1 commit
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
4c47d33
Adjust task handling in Worker to allow retrieving multiple locked ta…
rroblf01 Nov 7, 2025
d44e455
Modify the get_locked method in DBTaskResultQuerySet to allow retriev…
rroblf01 Nov 7, 2025
0f647bc
Adjust the query number assertions in the DatabaseBackendWorkerTestCa…
rroblf01 Nov 7, 2025
59eee96
Optimize task handling in the Worker and adjust query assertions in D…
rroblf01 Nov 7, 2025
b938687
Corrects task verification in the Worker's run method to handle multi…
rroblf01 Nov 7, 2025
cc282f2
Añade soporte para múltiples hilos en el Worker, permitiendo la ejecu…
rroblf01 Nov 7, 2025
75b092e
Adjust the query number assertions in DatabaseBackendWorkerTestCase t…
rroblf01 Nov 7, 2025
88a337a
Refactor test_repeat_ctrl_c to improve signal handling and process te…
rroblf01 Nov 7, 2025
9537b1d
Refactor Worker to support parallel task execution with ThreadPoolExe…
rroblf01 Nov 14, 2025
8d3c31f
Actualiza el tipo del argumento --max-threads en el comando Worker pa…
rroblf01 Nov 14, 2025
fa7ee12
Rename MAX_THREADS to DEFAULT_THREADS to improve clarity and consiste…
rroblf01 Nov 14, 2025
0eb3b19
Refactor run_parallel method to use threading instead of ThreadPoolEx…
rroblf01 Nov 18, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Adjust the query number assertions in the DatabaseBackendWorkerTestCa… 
…se and DatabaseTaskResultTestCase tests
  • Loading branch information
@rroblf01
rroblf01 committed Nov 7, 2025
commit 0f647bc60b02e25ab022c0f98a2f48b1720d45e1
38 changes: 20 additions & 18 deletions tests/tests/test_database_backend.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -559,7 +559,7 @@ def test_run_enqueued_task(self) -> None:

self.assertEqual(result.status, TaskResultStatus.READY)

with self.assertNumQueries(9 if connection.vendor == "mysql" else 8):
with self.assertNumQueries(9 if connection.vendor == "mysql" else 11):
self.run_worker()

self.assertEqual(result.status, TaskResultStatus.READY)
Expand All @@ -582,28 +582,28 @@ def test_batch_processes_all_tasks(self) -> None:

self.assertEqual(DBTaskResult.objects.ready().count(), 4)

with self.assertNumQueries(27 if connection.vendor == "mysql" else 23):
with self.assertNumQueries(27 if connection.vendor == "mysql" else 17):
self.run_worker()

self.assertEqual(DBTaskResult.objects.ready().count(), 0)
self.assertEqual(DBTaskResult.objects.succeeded().count(), 3)
self.assertEqual(DBTaskResult.objects.failed().count(), 1)

def test_no_tasks(self) -> None:
with self.assertNumQueries(3):
with self.assertNumQueries(5):
self.run_worker()

def test_doesnt_process_different_queue(self) -> None:
result = test_tasks.noop_task.using(queue_name="queue-1").enqueue()

self.assertEqual(DBTaskResult.objects.ready().count(), 1)

with self.assertNumQueries(3):
with self.assertNumQueries(5):
self.run_worker()

self.assertEqual(DBTaskResult.objects.ready().count(), 1)

with self.assertNumQueries(9 if connection.vendor == "mysql" else 8):
with self.assertNumQueries(9 if connection.vendor == "mysql" else 11):
self.run_worker(queue_name=result.task.queue_name)

self.assertEqual(DBTaskResult.objects.ready().count(), 0)
Expand All @@ -613,12 +613,12 @@ def test_process_all_queues(self) -> None:

self.assertEqual(DBTaskResult.objects.ready().count(), 1)

with self.assertNumQueries(3):
with self.assertNumQueries(5):
self.run_worker()

self.assertEqual(DBTaskResult.objects.ready().count(), 1)

with self.assertNumQueries(9 if connection.vendor == "mysql" else 8):
with self.assertNumQueries(9 if connection.vendor == "mysql" else 11):
self.run_worker(queue_name="*")

self.assertEqual(DBTaskResult.objects.ready().count(), 0)
Expand All @@ -627,7 +627,7 @@ def test_failing_task(self) -> None:
result = test_tasks.failing_task_value_error.enqueue()
self.assertEqual(DBTaskResult.objects.ready().count(), 1)

with self.assertNumQueries(9 if connection.vendor == "mysql" else 8):
with self.assertNumQueries(9 if connection.vendor == "mysql" else 11):
self.run_worker()

self.assertEqual(result.status, TaskResultStatus.READY)
Expand Down Expand Up @@ -656,7 +656,7 @@ def test_complex_exception(self) -> None:
result = test_tasks.complex_exception.enqueue()
self.assertEqual(DBTaskResult.objects.ready().count(), 1)

with self.assertNumQueries(9 if connection.vendor == "mysql" else 8):
with self.assertNumQueries(9 if connection.vendor == "mysql" else 11):
self.run_worker()

self.assertEqual(result.status, TaskResultStatus.READY)
Expand Down Expand Up @@ -701,12 +701,12 @@ def test_doesnt_process_different_backend(self) -> None:

self.assertEqual(DBTaskResult.objects.ready().count(), 1)

with self.assertNumQueries(3):
with self.assertNumQueries(5):
self.run_worker(backend_name="dummy")

self.assertEqual(DBTaskResult.objects.ready().count(), 1)

with self.assertNumQueries(9 if connection.vendor == "mysql" else 8):
with self.assertNumQueries(9 if connection.vendor == "mysql" else 11):
self.run_worker(backend_name=result.backend)

self.assertEqual(DBTaskResult.objects.ready().count(), 0)
Expand Down Expand Up @@ -794,7 +794,7 @@ def test_run_after(self) -> None:
self.assertEqual(DBTaskResult.objects.count(), 1)
self.assertEqual(DBTaskResult.objects.ready().count(), 0)

with self.assertNumQueries(3):
with self.assertNumQueries(5):
self.run_worker()

self.assertEqual(DBTaskResult.objects.count(), 1)
Expand All @@ -805,7 +805,7 @@ def test_run_after(self) -> None:

self.assertEqual(DBTaskResult.objects.ready().count(), 1)

with self.assertNumQueries(9 if connection.vendor == "mysql" else 8):
with self.assertNumQueries(9 if connection.vendor == "mysql" else 11):
self.run_worker()

self.assertEqual(DBTaskResult.objects.ready().count(), 0)
Expand Down Expand Up @@ -1055,7 +1055,7 @@ def test_locks_tasks_sqlite(self) -> None:
result = test_tasks.noop_task.enqueue()

with exclusive_transaction():
locked_result = DBTaskResult.objects.get_locked()
locked_result = DBTaskResult.objects.get_locked().first()

self.assertEqual(result.id, str(locked_result.id)) # type:ignore[union-attr]

Expand Down Expand Up @@ -1115,9 +1115,11 @@ def test_locks_tasks_filtered_sqlite(self) -> None:
test_tasks.noop_task.enqueue()

with exclusive_transaction():
locked_result = DBTaskResult.objects.filter(
priority=result.task.priority
).get_locked()
locked_result = (
DBTaskResult.objects.filter(priority=result.task.priority)
.get_locked()
.first()
)

self.assertEqual(result.id, str(locked_result.id))

Expand All @@ -1134,7 +1136,7 @@ def test_locks_tasks_filtered_sqlite(self) -> None:
@exclusive_transaction()
def test_lock_no_rows(self) -> None:
self.assertEqual(DBTaskResult.objects.count(), 0)
self.assertIsNone(DBTaskResult.objects.all().get_locked())
self.assertEqual(DBTaskResult.objects.all().get_locked().count(), 0)

@skipIf(connection.vendor == "sqlite", "SQLite handles locks differently")
def test_get_locked_with_locked_rows(self) -> None:
Expand Down