(PC-34345)[API] setup celery tasks for mails

api/pyproject.toml

@@ -16,6 +16,7 @@ dependencies = [
 "boto3==1.36.12",
 "boussole==2.1.3",
 "brevo-python==1.1.2",
+"celery[redis]==5.4.0",
 "click-option-group==0.5.6",
 "click==8.1.8",
 "clickhouse-sqlalchemy==0.2.7",
@@ -80,7 +81,7 @@ dependencies = [
 "wtforms-sqlalchemy==0.4.2",
 "wtforms==3.2.1",
 "xlsxwriter==3.2.2",
-"zeep==4.3.1",
+"zeep==4.3.1"
 ]
 
 [tool.poetry.group.dev]

api/src/pcapi/celery_tasks/celery.py

@@ -0,0 +1,19 @@
+from celery import Celery
+from celery import Task
+from flask import Flask
+
+
+def celery_init_app(app: Flask) -> Celery:
+    # pylint: disable=abstract-method
+    # Should be ok to only override the __call__ method
+    # https://docs.celeryq.dev/en/stable/reference/celery.app.task.html#celery.app.task.Task
+    class FlaskTask(Task):
+        def __call__(self, *args: object, **kwargs: object) -> object:
+            with app.app_context():
+                return self.run(*args, **kwargs)
+
+    celery_app = Celery(app.name, task_cls=FlaskTask)
+    celery_app.config_from_object(app.config["CELERY"])
+    celery_app.set_default()
+    app.extensions["celery"] = celery_app
+    return celery_app

api/src/pcapi/celery_tasks/celery_worker.py

@@ -0,0 +1,4 @@
+from pcapi.flask_app import app as flask_app
+
+
+celery_app = flask_app.extensions["celery"]

api/src/pcapi/celery_tasks/sendinblue.py

@@ -0,0 +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: dict) -> None:
+    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: 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: 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})

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

@@ -15,11 +15,13 @@
 from brevo_python.rest import ApiException as SendinblueApiException
 
 from pcapi import settings
+from pcapi.celery_tasks.sendinblue import update_contact_attributes_task_celery
 from pcapi.core import mails as mails_api
 from pcapi.core.cultural_survey import models as cultural_survey_models
 from pcapi.core.external.attributes import models as attributes_models
 import pcapi.core.users.models as users_models
-from pcapi.tasks.sendinblue_tasks import update_contact_attributes_task
+from pcapi.models.feature import FeatureToggle
+from pcapi.tasks.sendinblue_tasks import update_contact_attributes_task_cloud_tasks
 from pcapi.tasks.serialization.sendinblue_tasks import UpdateSendinblueContactRequest
 
 
@@ -121,9 +123,15 @@ def update_contact_email(user: users_models.User, old_email: str, new_email: str
     )
 
     if asynchronous:
-        update_contact_attributes_task.delay(contact_request)
+        if FeatureToggle.WIP_ASYNCHRONOUS_CELERY_TASKS.is_active():
+            update_contact_attributes_task_celery.delay(contact_request.dict())
+        else:
+            update_contact_attributes_task_cloud_tasks.delay(contact_request)
     else:
-        update_contact_attributes_task(contact_request)
+        if FeatureToggle.WIP_ASYNCHRONOUS_CELERY_TASKS.is_active():
+            update_contact_attributes_task_celery(contact_request.dict())
+        else:
+            update_contact_attributes_task_cloud_tasks(contact_request)
 
 
 def update_contact_attributes(
@@ -151,7 +159,10 @@ def update_contact_attributes(
     )
 
     if asynchronous:
-        update_contact_attributes_task.delay(contact_request)
+        if FeatureToggle.WIP_ASYNCHRONOUS_CELERY_TASKS.is_active():
+            update_contact_attributes_task_celery.delay(contact_request.dict())
+        else:
+            update_contact_attributes_task_cloud_tasks.delay(contact_request)
     else:
         make_update_request(contact_request)