[flows] allow to @on_flow_request callbacks to return or raise `Flo…

…wResponseError` subclasses

docs/source/content/flows/flow_types.rst

@@ -6,6 +6,8 @@ Flow Types
 .. autoclass:: FlowRequest()
     :members: has_error, is_health_check
 
+.. autoclass:: FlowRequestActionType()
+
 .. autoclass:: FlowResponse()
 
 .. autoclass:: FlowCategory()
@@ -21,6 +23,10 @@ Flow Types
 
 .. autoclass:: FlowAsset()
 
+.. autoclass:: FlowTokenNoLongerValid()
+
+.. autoclass:: FlowRequestSignatureAuthenticationFailed()
+
 .. currentmodule:: pywa.utils
 
 .. autoclass:: FlowRequestDecryptor()

pywa/types/flows.py

@@ -27,6 +27,9 @@
     "FlowCompletion",
     "FlowRequest",
     "FlowResponse",
+    "FlowRequestCannotBeDecrypted",
+    "FlowRequestSignatureAuthenticationFailed",
+    "FlowTokenNoLongerValid",
     "FlowCategory",
     "FlowDetails",
     "FlowStatus",
@@ -235,6 +238,53 @@ def to_dict(self) -> dict[str, str | dict]:
         }
 
 
+class FlowResponseError(Exception):
+    """Base class for all flow response errors"""
+
+    status_code: int
+
+
+class FlowRequestCannotBeDecrypted(FlowResponseError):
+    """
+    - The payload cannot be decrypted due to a private key being updated by your business.
+
+    - This error is returned automatically by pywa when the request cannot be decrypted.
+      The exception from the decryption function will still be logged.
+
+    - The WhatsApp client will re-fetch a public key and re-send the request.
+      If the request fails, an error modal appears with an acknowledge button which directs the user back
+      to your chat thread.
+    """
+
+    status_code = 421
+
+
+class FlowRequestSignatureAuthenticationFailed(FlowResponseError):
+    """
+    This exception need to be returned or raised from the flow endpoint callback when the request signature authentication fails.
+
+    - A generic error will be shown on the client.
+    """
+
+    status_code = 432
+
+
+class FlowTokenNoLongerValid(FlowResponseError):
+    """
+    This exception need to be returned or raised from the flow endpoint callback when the Flow token is no longer valid.
+
+    - The layout will be closed and the ``FlowButton`` will be disabled for the user.
+      You can send a new message to the user generating a new Flow token.
+      This action may be used to prevent users from initiating the same Flow again.
+    - You are able to set an error message to display to the user. e.g. “The order has already been placed”
+    """
+
+    status_code = 427
+
+    def __init__(self, error_message: str):
+        self.error_message = error_message
+
+
 class FlowStatus(utils.StrEnum):
     """
     The status of the flow

pywa/webhook.py

@@ -22,6 +22,11 @@
 )
 from pywa.types import MessageType, FlowRequest, FlowResponse
 from pywa.types.base_update import BaseUpdate, StopHandling  # noqa
+from pywa.types.flows import (
+    FlowRequestCannotBeDecrypted,
+    FlowResponseError,  # noqa
+    FlowTokenNoLongerValid,
+)
 from pywa.utils import FastAPI, Flask
 
 if TYPE_CHECKING:
@@ -219,6 +224,7 @@ def _register_flow_endpoint_callback(
         request_decryptor: utils.FlowRequestDecryptor | None,
         response_encryptor: utils.FlowResponseEncryptor | None,
     ) -> None:
+        """Internal function to register a flow endpoint callback."""
         if self._server is None:
             raise ValueError(
                 "You must initialize the WhatsApp client with an web server"
@@ -241,29 +247,45 @@ def _register_flow_endpoint_callback(
                 "The flow endpoint cannot be the same as the webhook endpoint."
             )
 
-        def flow_endpoint(payload: dict) -> str:
-            decrypted_request, aes_key, iv = (
-                request_decryptor or self._flows_request_decryptor
-            )(
-                payload["encrypted_flow_data"],
-                payload["encrypted_aes_key"],
-                payload["initial_vector"],
-                private_key or self._private_key,
-                private_key_password or self._private_key_password,
-            )
-            if handle_health_check and decrypted_request["action"] == "ping":
-                return (response_encryptor or self._flows_response_encryptor)(
-                    {
-                        "version": decrypted_request["version"],
-                        "data": {"status": "active"},
-                    },
-                    aes_key,
-                    iv,
+        def flow_endpoint(payload: dict) -> tuple[str, int]:
+            """The actual registered endpoint callback. returns response, status code"""
+            try:
+                decrypted_request, aes_key, iv = (
+                    request_decryptor or self._flows_request_decryptor
+                )(
+                    payload["encrypted_flow_data"],
+                    payload["encrypted_aes_key"],
+                    payload["initial_vector"],
+                    private_key or self._private_key,
+                    private_key_password or self._private_key_password,
                 )
+                if handle_health_check and decrypted_request["action"] == "ping":
+                    return (response_encryptor or self._flows_response_encryptor)(
+                        {
+                            "version": decrypted_request["version"],
+                            "data": {"status": "active"},
+                        },
+                        aes_key,
+                        iv,
+                    ), 200
+            except Exception as e:
+                _logger.exception(e)
+                return "Decryption failed", FlowRequestCannotBeDecrypted.status_code
             request = FlowRequest.from_dict(
                 data=decrypted_request, raw_encrypted=payload
             )
-            response = callback(self, request)
+            try:
+                response = callback(self, request)
+                if isinstance(response, FlowResponseError):
+                    raise response
+            except FlowTokenNoLongerValid as e:
+                return (
+                    """{"error_msg": %s}""" % e.error_message,
+                    FlowTokenNoLongerValid.status_code,
+                )
+            except FlowResponseError as e:
+                return e.__class__.__name__, e.status_code
+
             if acknowledge_errors and request.has_error:
                 return (response_encryptor or self._flows_response_encryptor)(
                     {
@@ -274,7 +296,7 @@ def flow_endpoint(payload: dict) -> str:
                     },
                     aes_key,
                     iv,
-                )
+                ), 200
             if not isinstance(response, (FlowResponse | dict)):
                 raise ValueError(
                     f"Flow endpoint callback must return a FlowResponse or dict, not {type(response)}"
@@ -283,23 +305,24 @@ def flow_endpoint(payload: dict) -> str:
                 response.to_dict() if isinstance(response, FlowResponse) else response,
                 aes_key,
                 iv,
-            )
+            ), 200
 
         if utils.is_flask_app(self._server):
             import flask
 
             @self._server.route(endpoint, methods=["POST"])
             @utils.rename_func(f"({endpoint})")
             def flow() -> tuple[str, int]:
-                return flow_endpoint(flask.request.json), 200
+                return flow_endpoint(flask.request.json)
 
         elif utils.is_fastapi_app(self._server):
             import fastapi
 
             @self._server.post(endpoint)
             @utils.rename_func(f"({endpoint})")
             def flow(payload: dict = fastapi.Body(...)):
+                response, status_code = flow_endpoint(payload)
                 return fastapi.Response(
-                    content=flow_endpoint(payload),
-                    status_code=200,
+                    content=response,
+                    status_code=status_code,
                 )