(PC-34345)[API] feat: use json serialization for celery tasks

api/src/pcapi/celery_tasks/sendinblue.py

@@ -1,26 +1,44 @@
+import logging
+
 from brevo_python.rest import ApiException as SendinblueApiException
 from celery import shared_task
+from pydantic import ValidationError
 
 from pcapi.core.external import sendinblue
 from pcapi.core.mails.transactional.send_transactional_email import send_transactional_email
 from pcapi.tasks.serialization.sendinblue_tasks import SendTransactionalEmailRequest
 from pcapi.tasks.serialization.sendinblue_tasks import UpdateSendinblueContactRequest
 
 
+logger = logging.getLogger(__name__)
+
+
 @shared_task(name="mails.tasks.update_contact_attributes", autoretry_for=(SendinblueApiException,), retry_backoff=True)
 def update_contact_attributes_task_celery(payload: UpdateSendinblueContactRequest) -> None:
-    sendinblue.make_update_request(payload)
+    try:
+        request = UpdateSendinblueContactRequest.parse_obj(payload)
+        sendinblue.make_update_request(request)
+    except ValidationError as exp:
+        logger.error("could not deserialize object", extra={"exception": exp})
 
 
 @shared_task(
     name="mails.tasks.send_transactional_email_primary", autoretry_for=(SendinblueApiException,), retry_backoff=True
 )
-def send_transactional_email_primary_task_celery(payload: SendTransactionalEmailRequest) -> None:
-    send_transactional_email(payload)
+def send_transactional_email_primary_task_celery(payload: dict) -> None:
+    try:
+        request = SendTransactionalEmailRequest.parse_obj(payload)
+        send_transactional_email(request)
+    except ValidationError as exp:
+        logger.error("could not deserialize object", extra={"exception": exp})
 
 
 @shared_task(
     name="mails.tasks.send_transactional_email_secondary", autoretry_for=(SendinblueApiException,), retry_backoff=True
 )
 def send_transactional_email_secondary_task_celery(payload: SendTransactionalEmailRequest) -> None:
-    send_transactional_email(payload)
+    try:
+        request = SendTransactionalEmailRequest.parse_obj(payload)
+        send_transactional_email(request)
+    except ValidationError as exp:
+        logger.error("could not deserialize object", extra={"exception": exp})

api/src/pcapi/core/external/sendinblue.py

@@ -124,12 +124,12 @@ def update_contact_email(user: users_models.User, old_email: str, new_email: str
 
     if asynchronous:
         if FeatureToggle.WIP_ASYNCHRONOUS_CELERY_TASKS.is_active():
-            update_contact_attributes_task_celery.delay(contact_request)
+            update_contact_attributes_task_celery.delay(contact_request.dict())
         else:
             update_contact_attributes_task_cloud_tasks.delay(contact_request)
     else:
         if FeatureToggle.WIP_ASYNCHRONOUS_CELERY_TASKS.is_active():
-            update_contact_attributes_task_celery(contact_request)
+            update_contact_attributes_task_celery(contact_request.dict())
         else:
             update_contact_attributes_task_cloud_tasks(contact_request)
 
@@ -160,9 +160,9 @@ def update_contact_attributes(
 
     if asynchronous:
         if FeatureToggle.WIP_ASYNCHRONOUS_CELERY_TASKS.is_active():
-            update_contact_attributes_task_celery.delay(contact_request)
+            update_contact_attributes_task_celery.delay(contact_request.dict())
         else:
-            update_contact_attributes_task_cloud_tasks.delay(contact_request)
+            update_contact_attributes_task_cloud_tasks.delay(contact_request.dict())
     else:
         make_update_request(contact_request)
 

api/src/pcapi/core/mails/backends/sendinblue.py

@@ -58,12 +58,12 @@ def send_mail(
             )
             if data.template.use_priority_queue:
                 if FeatureToggle.WIP_ASYNCHRONOUS_CELERY_TASKS.is_active():
-                    send_transactional_email_primary_task_celery.delay(payload)
+                    send_transactional_email_primary_task_celery.delay(payload.dict())
                 else:
                     send_transactional_email_primary_task_cloud_tasks.delay(payload)
             else:
                 if FeatureToggle.WIP_ASYNCHRONOUS_CELERY_TASKS.is_active():
-                    send_transactional_email_secondary_task_celery.delay(payload)
+                    send_transactional_email_secondary_task_celery.delay(payload.dict())
                 else:
                     send_transactional_email_secondary_task_cloud_tasks.delay(payload)
 
@@ -81,7 +81,7 @@ def send_mail(
                 tags=None,
             )
             if FeatureToggle.WIP_ASYNCHRONOUS_CELERY_TASKS.is_active():
-                send_transactional_email_secondary_task_celery.delay(payload)
+                send_transactional_email_secondary_task_celery.delay(payload.dict())
             else:
                 send_transactional_email_secondary_task_cloud_tasks.delay(payload)
 

api/src/pcapi/flask_app.py

@@ -173,11 +173,6 @@ def add_security_headers(response: flask.wrappers.Response) -> flask.wrappers.Re
         broker_url=settings.REDIS_URL,
         task_acks_late=True,
         task_reject_on_worker_lost=True,
-        # Pickle seems the best pick since we don't support
-        # anything other than python https://docs.celeryq.dev/en/latest/userguide/calling.html#serializers
-        task_serializer="pickle",
-        result_serializer="pickle",
-        accept_content=["pickle"],
         task_routes={
             "mails.tasks.*": {"queue": "mails"},
         },

api/tests/conftest.py

@@ -26,6 +26,7 @@
 import requests_mock
 import sqlalchemy as sa
 
+from pcapi.celery_tasks.celery import celery_init_app
 import pcapi.core.educational.testing as adage_api_testing
 import pcapi.core.mails.testing as mails_testing
 import pcapi.core.object_storage.testing as object_storage_testing
@@ -123,6 +124,15 @@ def build_main_app():
     # But in some tests, there are more recursions than the default accepted number (1000)
     sys.setrecursionlimit(3000)
 
+    app.config.from_mapping(
+        CELERY=dict(
+            # For testing, tasks are run locally
+            task_always_eager=True,
+        ),
+    )
+
+    celery_init_app(app)
+
     with app.app_context():
         app.config["TESTING"] = True
 

api/tests/core/mails/mails_celery_test.py

@@ -46,7 +46,7 @@ def _get_backend_for_test(self):
             "[email protected]",
         ]
     )
-    @patch("pcapi.core.mails.backends.sendinblue.send_transactional_email_secondary_task_celery.delay")
+    @patch("pcapi.celery_tasks.sendinblue.send_transactional_email")
     def test_send_mail(self, mock_send_transactional_email_secondary_task_celery):
         backend = self._get_backend_for_test()
         backend(use_pro_subaccount=False).send_mail(
@@ -64,7 +64,7 @@ def test_send_mail(self, mock_send_transactional_email_secondary_task_celery):
         assert task_param.reply_to == self.expected_sent_data.reply_to
         assert task_param.enable_unsubscribe == self.expected_sent_data.enable_unsubscribe
 
-    @patch("pcapi.core.mails.backends.sendinblue.send_transactional_email_secondary_task_celery.delay")
+    @patch("pcapi.celery_tasks.sendinblue.send_transactional_email")
     def test_send_mail_with_no_sender(self, mock_send_transactional_email_secondary_task_celery):
         self.mock_template = models.TemplatePro(
             id_prod=1,
@@ -162,7 +162,7 @@ def _get_backend_for_test(self):
         return import_string("pcapi.core.mails.backends.sendinblue.ToDevSendinblueBackend")
 
     @pytest.mark.settings(WHITELISTED_EMAIL_RECIPIENTS=["[email protected]"])
-    @patch("pcapi.core.mails.backends.sendinblue.send_transactional_email_secondary_task_celery.delay")
+    @patch("pcapi.celery_tasks.sendinblue.send_transactional_email")
     def test_send_mail_to_dev(self, mock_send_transactional_email_secondary_task_celery):
         backend = self._get_backend_for_test()
         backend(use_pro_subaccount=False).send_mail(
@@ -179,7 +179,7 @@ def test_send_mail_to_dev(self, mock_send_transactional_email_secondary_task_cel
         assert task_param.sender == self.expected_sent_data_to_dev.sender
         assert task_param.reply_to == self.expected_sent_data_to_dev.reply_to
 
-    @patch("pcapi.core.mails.backends.sendinblue.send_transactional_email_secondary_task_celery.delay")
+    @patch("pcapi.celery_tasks.sendinblue.send_transactional_email")
     def test_send_mail_test_user(self, mock_send_transactional_email_secondary_task_celery):
         users_factories.UserFactory(email=self.recipients[0], roles=[users_models.UserRole.TEST])
 
@@ -195,7 +195,7 @@ def test_send_mail_test_user(self, mock_send_transactional_email_secondary_task_
         ["[email protected]", "[email protected]"],
     )
     @pytest.mark.settings(WHITELISTED_EMAIL_RECIPIENTS=["[email protected]", "*@passculture-test.app"])
-    @patch("pcapi.core.mails.backends.sendinblue.send_transactional_email_secondary_task_celery.delay")
+    @patch("pcapi.celery_tasks.sendinblue.send_transactional_email")
     def test_send_mail_whitelisted(self, mock_send_transactional_email_secondary_task_celery, recipient):
         backend = self._get_backend_for_test()
         backend(use_pro_subaccount=False).send_mail(
@@ -207,7 +207,7 @@ def test_send_mail_whitelisted(self, mock_send_transactional_email_secondary_tas
         assert list(task_param.recipients) == [recipient]
 
     @pytest.mark.settings(IS_STAGING=True, IS_E2E_TESTS=True, END_TO_END_TESTS_EMAIL_ADDRESS="[email protected]")
-    @patch("pcapi.core.mails.backends.sendinblue.send_transactional_email_secondary_task_celery.delay")
+    @patch("pcapi.celery_tasks.sendinblue.send_transactional_email")
     def test_send_mail_whitelisted_qa_staging(self, mock_send_transactional_email_secondary_task_celery):
         recipient = "[email protected]"
         users_factories.UserFactory(email=recipient)
@@ -220,7 +220,7 @@ def test_send_mail_whitelisted_qa_staging(self, mock_send_transactional_email_se
         assert list(task_param.recipients) == [recipient]
 
     @pytest.mark.settings(IS_TESTING=True, IS_E2E_TESTS=True, END_TO_END_TESTS_EMAIL_ADDRESS="[email protected]")
-    @patch("pcapi.core.mails.backends.sendinblue.send_transactional_email_secondary_task_celery.delay")
+    @patch("pcapi.celery_tasks.sendinblue.send_transactional_email")
     def test_send_mail_whitelisted_qa_testing(
         self, mock_send_transactional_email_secondary_task_celery, recipient="[email protected]"
     ):
@@ -237,7 +237,7 @@ def test_send_mail_whitelisted_qa_testing(
 @pytest.mark.features(WIP_ASYNCHRONOUS_CELERY_TASKS=True)
 class SendTest:
     @pytest.mark.settings(IS_TESTING=True, EMAIL_BACKEND="pcapi.core.mails.backends.sendinblue.ToDevSendinblueBackend")
-    @patch("pcapi.core.mails.backends.sendinblue.send_transactional_email_secondary_task_celery.delay")
+    @patch("pcapi.celery_tasks.sendinblue.send_transactional_email")
     def test_send_to_ehp_false_in_testing(self, mock_send_transactional_email_secondary_task_celery, caplog):
         mock_template_send_ehp_false = models.Template(
             id_prod=11, id_not_prod=12, tags=["some", "stuff"], send_to_ehp=False
@@ -259,7 +259,7 @@ def test_send_to_ehp_false_in_testing(self, mock_send_transactional_email_second
         )
 
     @pytest.mark.settings(IS_TESTING=True, EMAIL_BACKEND="pcapi.core.mails.backends.sendinblue.ToDevSendinblueBackend")
-    @patch("pcapi.core.mails.backends.sendinblue.send_transactional_email_secondary_task_celery.delay")
+    @patch("pcapi.celery_tasks.sendinblue.send_transactional_email")
     def test_send_to_ehp_true_in_testing(self, mock_send_transactional_email_secondary_task_celery, caplog):
         mock_template_send_ehp_true = models.Template(
             id_prod=11, id_not_prod=12, tags=["some", "stuff"], send_to_ehp=True
@@ -282,7 +282,7 @@ def test_send_to_ehp_true_in_testing(self, mock_send_transactional_email_seconda
         ],
     )
     @pytest.mark.settings(EMAIL_BACKEND="pcapi.core.mails.backends.sendinblue.SendinblueBackend")
-    @patch("pcapi.core.mails.backends.sendinblue.send_transactional_email_secondary_task_celery.delay")
+    @patch("pcapi.celery_tasks.sendinblue.send_transactional_email")
     def test_send_mail_to_pro_with_FF(
         self,
         mock_send_transactional_email_secondary_task_celery,
@@ -313,7 +313,7 @@ def test_send_mail_to_pro_with_FF(
         ],
     )
     @pytest.mark.settings(IS_TESTING=True, EMAIL_BACKEND="pcapi.core.mails.backends.sendinblue.SendinblueBackend")
-    @patch("pcapi.core.mails.backends.sendinblue.send_transactional_email_secondary_task_celery.delay")
+    @patch("pcapi.celery_tasks.sendinblue.send_transactional_email")
     def test_send_mail_to_pro_with_FF_in_ehp(
         self,
         mock_send_transactional_email_secondary_task_celery,