(PC-38926)[API] fix: remove batching from Ubble async task

Ubble async task has its own rate limiting queue that should be a bit
smaller than the Ubble rate limiting queue. To make sure that the global
Ubble rate limiting queue does not overflow, then the Ubble async task
should treat ids one by one instead of by batches.

Author: dnguyen-pass

api/src/pcapi/core/subscription/api.py

@@ -632,6 +632,7 @@ def _update_fraud_check_eligibility_with_history(
         eligibilityType=eligibility,
     )
     db.session.add(new_fraud_check)
+    db.session.flush()
 
     # Cancel the old fraud check
     fraud_check.status = subscription_models.FraudCheckStatus.CANCELED

api/src/pcapi/core/subscription/ubble/api.py

@@ -188,8 +188,8 @@ def start_ubble_workflow(
         content = _reattempt_identity_verification(ubble_fraud_check, first_name, last_name, redirect_url, webhook_url)
     except ubble.UbbleConflictError:
         # resync the identity verification
-        ubble_tasks.update_ubble_workflow_task(
-            ubble_schemas.UpdateWorkflowPayload(beneficiary_fraud_check_ids=[ubble_fraud_check.id])
+        ubble_tasks.update_ubble_workflow_task.delay(
+            payload=ubble_schemas.UpdateWorkflowPayload(beneficiary_fraud_check_id=ubble_fraud_check.id).model_dump()
         )
 
         raise
@@ -450,16 +450,12 @@ def recover_pending_ubble_applications(dry_run: bool = True) -> None:
     pending_ubble_application_counter = 0
     for pending_ubble_application_fraud_checks in _get_pending_fraud_checks_pages():
         pending_ubble_application_counter += len(pending_ubble_application_fraud_checks)
-        if FeatureToggle.WIP_ASYNCHRONOUS_CELERY_UBBLE.is_active():
-            ubble_tasks.update_ubble_workflow_task(
-                ubble_schemas.UpdateWorkflowPayload(
-                    beneficiary_fraud_check_ids=[
-                        fraud_check.id for fraud_check in pending_ubble_application_fraud_checks
-                    ]
+        for fraud_check in pending_ubble_application_fraud_checks:
+            if FeatureToggle.WIP_ASYNCHRONOUS_CELERY_UBBLE.is_active():
+                ubble_tasks.update_ubble_workflow_task.delay(
+                    payload=ubble_schemas.UpdateWorkflowPayload(beneficiary_fraud_check_id=fraud_check.id).model_dump()
                 )
-            )
-        else:
-            for fraud_check in pending_ubble_application_fraud_checks:
+            else:
                 try:
                     with atomic():
                         update_ubble_workflow(fraud_check)

api/src/pcapi/core/subscription/ubble/schemas.py

@@ -129,4 +129,4 @@ def get_id_piece_number(self) -> str | None:
 
 
 class UpdateWorkflowPayload(BaseModelV2):
-    beneficiary_fraud_check_ids: list[int]
+    beneficiary_fraud_check_id: int

api/src/pcapi/core/subscription/ubble/tasks.py

@@ -17,7 +17,7 @@
 logger = logging.getLogger(__name__)
 
 # always leave some bandwidth for manual Ubble call through native app
-UBBLE_TASK_RATE_LIMIT = int(settings.UBBLE_RATE_LIMIT * 0.8)
+UBBLE_TASK_RATE_LIMIT = int(settings.UBBLE_RATE_LIMIT * 0.9)
 
 
 @celery_async_task(
@@ -30,31 +30,23 @@
 def update_ubble_workflow_task(payload: ubble_schemas.UpdateWorkflowPayload) -> None:
     fraud_check_stmt = (
         sa.select(subscription_models.BeneficiaryFraudCheck)
-        .where(subscription_models.BeneficiaryFraudCheck.id.in_(payload.beneficiary_fraud_check_ids))
+        .where(subscription_models.BeneficiaryFraudCheck.id == payload.beneficiary_fraud_check_id)
         .options(
             sa.orm.joinedload(subscription_models.BeneficiaryFraudCheck.user)
             .selectinload(users_models.User.deposits)
             .selectinload(finance_models.Deposit.recredits)
         )
     )
-    fraud_checks_to_update = db.session.scalars(fraud_check_stmt).all()
+    fraud_check = db.session.scalars(fraud_check_stmt).one()
 
-    for fraud_check in fraud_checks_to_update:
-        try:
-            with atomic():
-                ubble_api.update_ubble_workflow(fraud_check)
-        except Exception as e:
-            logger.error(
-                "Error while updating pending ubble application",
-                extra={"fraud_check_id": fraud_check.id, "ubble_id": fraud_check.thirdPartyId, "exc": str(e)},
-            )
-            continue
+    with atomic():
+        ubble_api.update_ubble_workflow(fraud_check)
 
-        if fraud_check.status == subscription_models.FraudCheckStatus.PENDING:
-            logger.error(
-                "Pending ubble application still pending after 12 hours. This is a problem on the Ubble side.",
-                extra={"fraud_check_id": fraud_check.id, "ubble_id": fraud_check.thirdPartyId},
-            )
+    if fraud_check.status == subscription_models.FraudCheckStatus.PENDING:
+        logger.error(
+            "Pending ubble application still pending after 12 hours. This is a problem on the Ubble side.",
+            extra={"fraud_check_id": fraud_check.id, "ubble_id": fraud_check.thirdPartyId},
+        )
 
 
 def update_ubble_workflow_if_needed(
@@ -68,16 +60,14 @@ def update_ubble_workflow_if_needed(
     """
     if status in ubble_api.PENDING_STATUSES:
         fraud_check.status = subscription_models.FraudCheckStatus.PENDING
-        db.session.commit()
         return
 
     if status in ubble_api.CANCELED_STATUSES:
         fraud_check.status = subscription_models.FraudCheckStatus.CANCELED
-        db.session.commit()
         return
 
     if status not in ubble_api.CONCLUSIVE_STATUSES:
         return
 
-    payload = ubble_schemas.UpdateWorkflowPayload(beneficiary_fraud_check_ids=[fraud_check.id])
-    update_ubble_workflow_task.delay(payload.dict())
+    payload = ubble_schemas.UpdateWorkflowPayload(beneficiary_fraud_check_id=fraud_check.id)
+    update_ubble_workflow_task.delay(payload.model_dump())

api/tests/core/subscription/ubble/test_api.py

@@ -1469,7 +1469,7 @@ def test_not_eligible(self):
 
 
 @pytest.mark.usefixtures("db_session")
-@patch("pcapi.core.subscription.ubble.api.update_ubble_workflow")
+@patch("pcapi.core.subscription.ubble.tasks.update_ubble_workflow_task.delay")
 def test_pending_and_created_fraud_checks_are_updated(update_ubble_workflow_mock):
     yesterday = datetime.date.today() - relativedelta(hours=13)
     created_fraud_check = BeneficiaryFraudCheckFactory(
@@ -1481,4 +1481,10 @@ def test_pending_and_created_fraud_checks_are_updated(update_ubble_workflow_mock
 
     ubble_subscription_api.recover_pending_ubble_applications()
 
-    update_ubble_workflow_mock.assert_has_calls([call(created_fraud_check), call(pending_fraud_check)], any_order=True)
+    update_ubble_workflow_mock.assert_has_calls(
+        [
+            call(payload={"beneficiary_fraud_check_id": created_fraud_check.id}),
+            call(payload={"beneficiary_fraud_check_id": pending_fraud_check.id}),
+        ],
+        any_order=True,
+    )