feat: allow discarding tasks in batch

README.md

@@ -56,7 +56,7 @@ MyTask.asap(x=10, y=3)  # run async (another instance will execute and print)
 MyTask.sync(x=10, y=5)  # run sync (the print happens right now)
 ```
 
-It's also possible to execute asynchronously, but not immediatelly:
+It's also possible to execute asynchronously, but not immediately:
 ```python
 MyTask.later(task_kwargs=dict(x=10, y=5), eta=3600)  # run async in 1 hour (int, timedelta and datetime are accepted)
 
@@ -89,7 +89,7 @@ class MyTask(Task):
         return "my-queue-name-here"
 ```
 
-### Throublehsooting
+### Troubleshooting
 
 When a task if failing in Cloud Tasks and you want to debug **locally** with the same data, 
 you can get the task ID from Cloud Task UI (the big number in the column NAME) and run the task locally with the same parameters with:
@@ -98,6 +98,32 @@ you can get the task ID from Cloud Task UI (the big number in the column NAME) a
 MyTask.debug(task_id="<the task number>")
 ```
 
+### Cleanup
+
+Google Cloud Tasks will automatically discard any jobs after the max-retries.
+
+If by any reason you need to discard jobs manually, you can provide the Task ID:
+
+
+```python
+MyTask.discard(task_id="<the task number>")
+```
+
+Or you can batch discard many tasks at once:
+
+
+```python
+MyTask.discard()
+```
+
+
+You can also provide `min_retries` parameter to filter the tasks that have retried at least some amount 
+(so tasks have some chance to execute):
+
+```python
+MyTask.discard(min_retries=5)
+```
+
 ## Periodic Task
 
 Tasks can be executed recurrently, using a crontab syntax.

django_cloud_tasks/tasks/task.py

@@ -311,6 +311,28 @@ def debug(cls, task_id: str):
         metadata = TaskMetadata.from_task_obj(task_obj=task_obj)
         return cls(metadata=metadata).run(**task_kwargs)
 
+    @classmethod
+    def discard(cls, task_id: str | None = None, min_retries: int = 0):
+        client = cls._get_tasks_client()
+        if task_id:
+            task_objects = [client.get_task(queue_name=cls.queue(), task_name=task_id)]
+        else:
+            task_objects = client.list_tasks(queue_name=cls.queue())
+
+        outputs = []
+        for task_obj in task_objects:
+            task_name = task_obj.http_request.url.rsplit("/", 1)[-1]
+            if task_name != cls.name():
+                continue
+
+            if task_obj.dispatch_count < min_retries:
+                continue
+
+            task_id = task_obj.name.split("/")[-1]
+            client.delete_task(queue_name=cls.queue(), task_name=task_id)
+            outputs.append(f"{task_name}/{task_id}")
+        return outputs
+
     @classmethod
     def name(cls) -> str:
         return str(cls)