fix: do not break when parsing headers from non-json messages

django_cloud_tasks/middleware/pubsub_headers_middleware.py

@@ -1,4 +1,5 @@
 import logging
+from json import JSONDecodeError
 from typing import Any
 
 from django.apps import apps
@@ -41,7 +42,11 @@ def is_subscriber_route(self, request) -> bool:
         return request.path == expected_url
 
     def extract_headers(self, request) -> dict[str, Any]:
-        message = Message.load(body=request.body)
+        try:
+            message = Message.load(body=request.body)
+        except JSONDecodeError:
+            logger.warning("Message received through PubSub is not a valid JSON. Ignoring PubSub headers feature.")
+            return {}
 
         headers = {}
         for key, value in message.attributes.items():

django_cloud_tasks/tasks/subscriber_task.py

@@ -1,5 +1,7 @@
 import abc
+import json
 from urllib.parse import urljoin
+from typing import Callable
 
 from cachetools.func import lru_cache
 from django.urls import reverse
@@ -21,6 +23,12 @@ class SubscriberTask(Task, abc.ABC):
     def run(self, content: dict, attributes: dict[str, str] | None = None):
         raise NotImplementedError()
 
+    @classmethod
+    def message_parser(cls) -> Callable:
+        # The callable used to parse the message content
+        # By default, we handle JSON messages
+        return json.loads
+
     @classmethod
     def set_up(cls):
         return cls._get_subscriber_client().create_or_update_subscription(

django_cloud_tasks/views.py

@@ -20,7 +20,7 @@ def post(self, request, task_name, *args, **kwargs):
             result = {"error": f"Task {task_name} not found"}
             return JsonResponse(status=404, data=result)
 
-        task_kwargs = self.parse_input(request=request)
+        task_kwargs = self.parse_input(request=request, task_class=task_class)
         task_metadata = self.parse_metadata(request=request)
         try:
             output = self.execute_task(task_class=task_class, task_metadata=task_metadata, task_kwargs=task_kwargs)
@@ -39,7 +39,7 @@ def get_task(self, name: str) -> Type[Task]:
     def execute_task(self, task_class: type[Task], task_metadata: TaskMetadata, task_kwargs: dict) -> Any:
         return task_class(metadata=task_metadata).process(**task_kwargs)
 
-    def parse_input(self, request) -> dict:
+    def parse_input(self, request, task_class: Type[Task]) -> dict:
         return deserialize(value=request.body)
 
     def parse_metadata(self, request) -> TaskMetadata:
@@ -48,8 +48,8 @@ def parse_metadata(self, request) -> TaskMetadata:
 
 # More info: https://cloud.google.com/pubsub/docs/push#receiving_messages
 class GoogleCloudSubscribeView(GoogleCloudTaskView):
-    def parse_input(self, request) -> dict:
-        message = Message.load(body=request.body)
+    def parse_input(self, request, task_class: Type[SubscriberTask]) -> dict:
+        message = Message.load(body=request.body, parser=task_class.message_parser())
         return {
             "content": message.data,
             "attributes": message.attributes,

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "django-google-cloud-tasks"
-version = "2.1.4"
+version = "2.1.5"
 description = "Async Tasks with HTTP endpoints"
 authors = ["Joao Daher <[email protected]>"]
 packages = [