Raises JSONDecodeError when I pass data using pickle.dumps method to PeriodicTask
I want to make feature to make posts at chosen time in Django REST Framework with Celery. I realised models, serializer, viewset, add settings related with Celery to Django. But I get error when I pass the data to task using pickle.dumps method:
models.py:
class Profile(models.Model):
class MaritalStatusChoices(models.TextChoices):
MARRIED = "MARRIED", _("Married")
DIVORCED = "DIVORCED", _("Divorced")
ACTIVELY_LOOKING = "ACTIVELY_LOOKING", _("Is actively looking")
HAS_FRIEND = "HAS_FRIEND", _("Has a friend")
NOT_SPECIFIED = "NOT_SPECIFIED", _("Not specified")
class GenderChoices(models.TextChoices):
MALE = "MALE", _("Male")
FEMALE = "FEMALE", _("Female")
GENDERLESS = "GENDERLESS", _("Genderless")
username = models.CharField(
max_length=150,
unique=True,
validators=(UnicodeUsernameValidator(),),
)
first_name = models.CharField(_("first name"), max_length=150)
last_name = models.CharField(_("last name"), max_length=150)
image = models.ImageField(
null=True,
blank=True,
upload_to=create_custom_image_path
)
bio = models.TextField(null=True, blank=True)
birth_date = models.DateField()
marital_status = models.CharField(
choices=MaritalStatusChoices,
default=MaritalStatusChoices.NOT_SPECIFIED
)
gender = models.CharField(choices=GenderChoices,)
created_at = models.DateField(auto_now_add=True)
user = models.OneToOneField(
settings.AUTH_USER_MODEL,
on_delete=models.CASCADE,
related_name="profile"
)
followings = models.ManyToManyField(
"Profile",
related_name="followers",
blank=True,
)
class Meta:
ordering = ("username",)
def __str__(self) -> str:
return f"{self.username} {self.first_name} {self.last_name}"
class Post(models.Model):
title = models.CharField(max_length=64)
content = models.TextField()
creation_date = models.DateTimeField(
blank=True,
default=timezone.now
)
hashtags = models.ManyToManyField(
Hashtag,
related_name="posts",
blank=True
)
image = models.ImageField(
null=True,
blank=True,
upload_to=create_custom_image_path
)
profile = models.ForeignKey(
Profile,
on_delete=models.CASCADE,
related_name="posts"
)
class Meta:
ordering = ("-creation_date",)
@property
def hashtags_display(self) -> str:
return ", ".join(str(hashtag) for hashtag in self.hashtags.all())
def __str__(self) -> str:
return f"{self.title} ({self.profile})"
serializers.py:
class CurrentUserProfileDefault(serializers.CurrentUserDefault):
def __call__(self, serializer_field: serializers.Field) -> models.Profile:
return super().__call__(serializer_field).profile
class PostSerializer(serializers.ModelSerializer):
profile = serializers.HiddenField(
default=CurrentUserProfileDefault()
)
class Meta:
model = models.Post
fields = (
"id",
"title",
"content",
"creation_date",
"hashtags",
"image",
"profile"
)
views.py:
class PostViewSet(viewsets.ModelViewSet):
queryset = models.Post.objects.prefetch_related(
"hashtags",
"likes",
"comments"
)
filterset_class = filters.PostFilterSet
pagination_class = pagination.PostPaginationSet
permission_classes = (HasProfileToPerformActionsOrReadOnly,)
def get_serializer_class(self) -> type[serializers.PostSerializer]:
if self.action in ("list", "retrieve"):
return serializers.PostListRetrieveSerializer
elif self.action in ("update", "partial_update"):
return serializers.PostUpdateSerializer
return serializers.PostSerializer
def create(self, request: Request, *args, **kwargs) -> Response:
if request.data["creation_date"]:
serializer = self.get_serializer(
data=request.data,
context={"request": request}
)
serializer.is_valid(raise_exception=True)
post_data = serializer.validated_data
title = post_data["title"]
creation_date = post_data.pop("creation_date")
schedule = ClockedSchedule.objects.create(
clocked_time=creation_date,
)
PeriodicTask.objects.create(
name=f"Create post {title} at {creation_date}",
task="content.tasks.create_post",
clocked=schedule,
args=[pickle.dumps(post_data)],
one_off=True,
)
response_data = {
"accepted": f"Post {title} will be created at {creation_date}"
}
return Response(
data=response_data,
status=status.HTTP_202_ACCEPTED
)
return super().create(request, *args, **kwargs)
tasks.py
from celery import shared_task
from content.models import Post
@shared_task(serializer="pickle")
def create_post(post_data: dict) -> None:
hashtags = post_data.pop("hashtags")
post = Post.objects.create(**post_data)
post.hashtags.set(hashtags)
post.save()
Error when I launch celery-beat:
Traceback (most recent call last):
File "D:\Programming\Python\Projects\social-media-api\.venv\Lib\site-packages\django_celery_beat\schedulers.py", line 69, in __init__
self.args = loads(model.args or '[]')
^^^^^^^^^^^^^^^^^^^^^^^^^
File "D:\Programming\Python\Projects\social-media-api\.venv\Lib\site-packages\kombu\utils\json.py", line 93, in loads
return _loads(s, object_hook=object_hook)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\Win\AppData\Local\Programs\Python\Python312\Lib\json\__init__.py", line 359, in loads
return cls(**kw).decode(s)
^^^^^^^^^^^^^^^^^^^
File "C:\Users\Win\AppData\Local\Programs\Python\Python312\Lib\json\decoder.py", line 337, in decode
obj, end = self.raw_decode(s, idx=_w(s, 0).end())
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "C:\Users\Win\AppData\Local\Programs\Python\Python312\Lib\json\decoder.py", line 355, in raw_decode
raise JSONDecodeError("Expecting value", s, err.value) from None
json.decoder.JSONDecodeError: Expecting value: line 1 column 2 (char 1)