Flow Types

Flow Types#

class pywa.types.flows.FlowRequest#

Represents a flow data exchange request. This request is sent to the flow endpoint when a user interacts with a flow and perform an Action that trigger a data exchange.

Read more at developers.facebook.com.

Variables:

version (str) – The version of the data exchange.
flow_token (str | None) – The flow token used to create the flow. None if action is FlowRequestActionType.PING.
action (pywa.types.flows.FlowRequestActionType) – The action that triggered the request.
screen (str | None) – The screen that triggered the request. None if action is FlowRequestActionType.PING.
data (dict[str, Any] | None) – The data sent from the screen. None if action is FlowRequestActionType.PING and optional if action is FlowRequestActionType.BACK or FlowRequestActionType.INIT.
raw (dict[str, Any]) – The raw data of the request.
raw_encrypted (dict[str, str]) – The raw-encrypted data of the request.

Create a response for this request.

A shortcut for initializing a FlowResponse with the same version as this request.

Example

>>> wa = WhatsApp(business_private_key="...", ...)
>>> @wa.on_flow_request("/my-flow-endpoint")
... def my_flow_endpoint(_: WhatsApp, req: FlowRequest) -> FlowResponse:
...     return req.respond(
...         screen="SCREEN_ID",
...         data={"key": "value"},
...     )

Parameters:

screen – The screen to display (if close_flow is False).
data – The data to send to the screen or to add to flow completion message (default to empty dict).
error_message – This will redirect the user to screen and will trigger a snackbar error with the error_message present (if close_flow is False).
close_flow – Whether to close the flow or just navigate to the screen.
flow_token – The flow token to close the flow (if close_flow is True).
version – The version of the data exchange (Default to the request version).

property has_error: bool#: Check if the request has an error. When True, if flow endpoint register with acknowledge_errors=True, pywa will acknowledge the error and ignore the response from the callback. The callback still be called.

property is_health_check: bool#: Check if the request is a health check. When True, if flow endpoint register with handle_health_check=True, pywa will not call the callback and will return a health check response.

class pywa.types.flows.FlowRequestActionType#

The type the action that triggered the FlowRequest.

Variables:

INIT – if the request is triggered when opening the flow (The FlowButton was sent with flow_action_type set to FlowActionType.DATA_EXCHANGE)
BACK – if the request is triggered when pressing back (The screen has refresh_on_back set to True)
DATA_EXCHANGE – if the request is triggered when submitting the screen (And the Action name is FlowActionType.DATA_EXCHANGE)
PING – if the request is triggered by a health check (Ignore this requests by leaving handle_health_check to True)
NAVIGATE – if the FlowButton sent with FlowActionType.NAVIGATE and the screen is not in the routing model (the request will contain an error)

class pywa.types.flows.FlowResponse#

Represents a flow data exchange response. This response is sent to the flow endpoint to determine the next screen to display or to close the flow. You should return this response from your flow endpoint callback.

Read more at developers.facebook.com.
Use the FlowRequest.respond() method to create a response for a request with the same version.

Example

>>> from pywa import WhatsApp
>>> from pywa.types.flows import FlowResponse

>>> wa = WhatsApp(business_private_key="...", ...)
>>> @wa.on_flow_request("/my-flow-endpoint")
... def my_flow_endpoint(flow_request: FlowRequest) -> FlowResponse:
...     return FlowResponse(
...         version=flow_request.version,
...         screen="SCREEN_ID",
...         data={"key": "value"},
...     )

Variables:

version (str) – The version of the data exchange (You can use the same version as the request (request.version)).
screen (str | None) – The screen to display (if the flow is not closed).
data (dict[str, Union[str, int, float, bool, dict, pywa.types.flows.DataSource, Iterable[str | int | float | bool | dict | pywa.types.flows.DataSource]]]) – The data to send to the screen or to add to flow completion message (default to empty dict).
error_message (str | None) – This will redirect the user to screen and will trigger a snackbar error with the error_message present (if the flow is not closed).
flow_token (str | None) – The flow token to close the flow (if close_flow is True).
close_flow (bool) – Whether to close the flow or just navigate to the screen.

class pywa.types.flows.FlowCategory#

The category of the flow

Variables:

SIGN_UP – Sign up
SIGN_IN – Sign in
APPOINTMENT_BOOKING – Appointment booking
LEAD_GENERATION – Lead generation
CONTACT_US – Contact us
CUSTOMER_SUPPORT – Customer support
SURVEY – Survey
OTHER – Other

class pywa.types.flows.FlowDetails#

Represents the details of a flow.

Read more at developers.facebook.com.

Variables:

id (str) – The ID of the flow.
name (str) – The name of the flow.
status (FlowStatus) – The status of the flow.
updated_at (datetime.datetime | None) – The last time the flow was updated (name, categories, endpoint_uri, json, etc.).
json_version (str | None) – The version of the flow JSON.
data_api_version (str | None) – The version to use during communication with the WhatsApp Flows Data Endpoint.
categories (tuple[FlowCategory, ...]) – The categories of the flow.
validation_errors (tuple[FlowValidationError, ...] | None) – The validation errors of the flow.
endpoint_uri (str | None) – The endpoint URI of the flow.
preview (FlowPreview | None) – The preview of the flow.
whatsapp_business_account (WhatsAppBusinessAccount | None) – The WhatsApp Business Account that owns the flow.
application (FacebookApplication | None) – The application that owns the flow.

publish() → bool#

Update the status of this flow to FlowStatus.PUBLISHED.

A shortcut for pywa.client.WhatsApp.publish_flow().

This action is not reversible.
The Flow and its assets become immutable once published.
To update the Flow after that, you must create a new Flow. You specify the existing Flow ID as the clone_flow_id parameter while creating to copy the existing flow.
You can publish your Flow once you have ensured that:
- All validation errors and publishing checks have been resolved.
- The Flow meets the design principles of WhatsApp Flows
- The Flow complies with WhatsApp Terms of Service, the WhatsApp Business Messaging Policy and, if applicable, the WhatsApp Commerce Policy

Returns:: Whether the flow was published.
Raises:: FlowPublishingError – If this flow has validation errors or not all publishing checks have been resolved.

delete() → bool#

When the flow is in FlowStatus.DRAFT status, you can delete it.

A shortcut for pywa.client.WhatsApp.delete_flow().

Returns:: Whether the flow was deleted.
Raises:: FlowDeletingError – If this flow is already published.

deprecate() → bool#

When the flow is in FlowStatus.PUBLISHED status, you can only deprecate it.

A shortcut for pywa.client.WhatsApp.deprecate_flow().

Returns:: Whether the flow was deprecated.
Raises:: FlowDeprecatingError – If this flow is not published or already deprecated.

get_assets() → tuple[pywa.types.flows.FlowAsset, ...]#

Get all assets attached to this flow.

A shortcut for pywa.client.WhatsApp.get_flow_assets().

Returns:: The assets of the flow.

Update the metadata of this flow.

A shortcut for pywa.client.WhatsApp.update_flow_metadata().

Parameters:

name – The name of the flow (optional).
categories – The new categories of the flow (optional).
endpoint_uri – The URL of the FlowJSON Endpoint. Starting from FlowJSON 3.0 this property should be specified only gere. Do not provide this field if you are cloning a FlowJSON with version below 3.0.
application_id – The ID of the Meta application which will be connected to the Flow. All the flows with endpoints need to have an Application connected to them.

Example

>>> from pywa.types.flows import FlowCategory
>>> wa = WhatsApp(business_account_id='1234567890', ...)
>>> my_flows = wa.get_flows()
>>> my_flows[0].update_metadata(
...     name='Feedback',
...     categories=[FlowCategory.SURVEY, FlowCategory.OTHER],
...     endpoint_uri='https://my-api-server/feedback_flow'
... )

Returns:: Whether the flow was updated.
Raises:: ValueError – If neither name, categories or endpoint_uri is provided.

Update the json of this flow.

A shortcut for pywa.client.WhatsApp.update_flow_json().

Parameters:: flow_json – The new json of the flow. Can be a FlowJSON object, dict, json str, json file path or json bytes.
Returns:: Whether the flow was updated.
Raises:: FlowUpdatingError – If the flow json is invalid or this flow is already published.

class pywa.types.flows.FlowStatus#

The status of the flow

Variables:

DRAFT – This is the initial status. The Flow is still under development. The Flow can only be sent with “mode”: “draft” for testing.
PUBLISHED – The Flow has been marked as published by the developer so now it can be sent to customers. This Flow cannot be deleted or updated afterwards.
DEPRECATED – The developer has marked the Flow as deprecated (since it cannot be deleted after publishing). This prevents sending and opening the Flow, to allow the developer to retire their endpoint. Deprecated Flows cannot be deleted or deprecated.
BLOCKED – Monitoring detected that the endpoint is unhealthy and set the status to Blocked. The Flow cannot be sent or opened in this state; the developer needs to fix the endpoint to get it back to Published state (more details in Flows Health and Monitoring).
THROTTLED –
Monitoring detected that the endpoint is unhealthy and set the status to Throttled. Flows with throttled status can be opened, however only 10 messages of the Flow could be sent per hour. The developer needs to fix the endpoint to get it back to the PUBLISHED state (more details in Flows Health and Monitoring on developers.facebook.com).

class pywa.types.flows.FlowPreview#

Represents the preview of a flow.

Variables:

url (str) – The URL to the preview.
expires_at (datetime.datetime) – The expiration date of the preview.

class pywa.types.flows.FlowValidationError#

Represents a validation error of a FlowJSON.

Variables:

error (str) – The error code.
error_type (str) – The type of the error.
message (str) – The error message.
line_start (int) – The start line of the error.
line_end (int) – The end line of the error.
column_start (int) – The start column of the error.
column_end (int) – The end column of the error.

class pywa.types.flows.FlowAsset#

Represents an asset in a flow.

Read more at developers.facebook.com.

Variables:

name (str) – The name of the asset (e.g. "flow.json").
type (str) – The type of the asset (e.g. "FLOW_JSON").
url (str) – The URL to the asset.

class pywa.types.flows.FlowResponseError#

Bases: Exception

Base class for all flow response errors

Subclass this exception to return or raise from the flow endpoint callback (@wa.on_flow_request).
Override the status_code attribute to set the status code of the response.
Override the body attribute to set the body of the response (optional).

class pywa.types.flows.FlowTokenNoLongerValid#

Bases: FlowResponseError

This exception need to be returned or raised from the flow endpoint callback when the Flow token is no longer valid.

Example

>>> from pywa.types.flows import FlowTokenNoLongerValid
>>> raise FlowTokenNoLongerValid(error_message='The order has already been placed')

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”

class pywa.types.flows.FlowRequestSignatureAuthenticationFailed#

Bases: 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.

pywa.utils.FlowRequestDecryptor#

Type hint for the function that decrypts the request from WhatsApp Flow.

All parameters need to be positional.
See default_flow_request_decryptor() source code for an example.

Parameters:

encrypted_flow_data_b64 (str) – encrypted flow data
encrypted_aes_key_b64 (str) – encrypted AES key
initial_vector_b64 (str) – initial vector
private_key (str) – private key
password (str) – password for the private key. Optional.

Returns:

tuple[dict, bytes, bytes] - decrypted_data (dict): decrypted data from the request - aes_key (bytes): AES key you should use to encrypt the response - iv (bytes): initial vector you should use to encrypt the response

pywa.utils.FlowResponseEncryptor#

Type hint for the function that encrypts the response to WhatsApp Flow.

All parameters need to be positional.
See default_flow_response_encryptor() source code for an example.

Parameters:

response (dict) – response to encrypt
aes_key (bytes) – AES key
iv (bytes) – initial vector

Returns:

encrypted response to send back to WhatsApp Flow

Return type:

encrypted_response (str)

pywa.utils.default_flow_request_decryptor(encrypted_flow_data_b64: str, encrypted_aes_key_b64: str, initial_vector_b64: str, private_key: str, password: str = None) → tuple[dict, bytes, bytes]#

The default global decryption function for decrypting data exchange requests from WhatsApp Flow.

This implementation follows the FlowRequestDecryptor type hint.
This implementation requires cryptography to be installed. To install it, run pip3 install 'pywa[cryptography]' or pip3 install cryptography.
This implementation was taken from the official documentation at developers.facebook.com.

Example

Set the default global decryptor (This is indeed the default):

>>> from pywa.utils import default_flow_request_decryptor
>>> from pywa import WhatsApp
>>> wa = WhatsApp(flows_request_decryptor=default_flow_request_decryptor, ...)

Set the decryptor for a specific flow:

>>> from pywa import WhatsApp
>>> from pywa.types.flows import FlowRequest, FlowResponse
>>> from pywa.utils import default_flow_request_decryptor
>>> wa = WhatsApp(...)
>>> @wa.on_flow_request("/sign-up-flow", request_decryptor=default_flow_request_decryptor)
... def on_sign_up_request(_: WhatsApp, flow: FlowRequest) -> FlowResponse | None: ...

pywa.utils.default_flow_response_encryptor(response: dict, aes_key: bytes, iv: bytes) → str#

The default global encryption function for encrypting data exchange responses to WhatsApp Flow.

This implementation follows the FlowResponseEncryptor type hint.
This implementation requires cryptography to be installed. To install it, run pip3 install 'pywa[cryptography]' or pip3 install cryptography.
This implementation was taken from the official documentation at developers.facebook.com.

Example

Set the default global encryptor (This is indeed the default):

>>> from pywa.utils import default_flow_response_encryptor
>>> from pywa import WhatsApp
>>> wa = WhatsApp(flows_response_encryptor=default_flow_response_encryptor, ...)

Set the encryptor for a specific flow:

>>> from pywa import WhatsApp
>>> from pywa.types.flows import FlowRequest, FlowResponse
>>> from pywa.utils import default_flow_response_encryptor
>>> wa = WhatsApp(...)
>>> @wa.on_flow_request("/sign-up-flow", response_encryptor=default_flow_response_encryptor)
... def on_sign_up_request(_: WhatsApp, flow: FlowRequest) -> FlowResponse | None: ...

pywa.utils.flow_request_media_decryptor_sync(encrypted_media: dict[str, str | dict[str, str]], dl_session: Session | Client | None = None) → tuple[str, str, bytes]#

Decrypt the encrypted media file from the flow request.

Read more at developers.facebook.com.
This implementation requires cryptography to be installed. To install it, run pip3 install 'pywa[cryptography]' or pip3 install cryptography.
This implementation is synchronous. Use this if your flow request callback is synchronous, otherwise use flow_request_media_decryptor_async().

Example

>>> from pywa import WhatsApp, utils, types
>>> wa = WhatsApp(...)
>>> @wa.on_flow_request("/media-upload")
... def on_media_upload_request(_: WhatsApp, flow: types.FlowRequest) -> types.FlowResponse | None:
...     encrypted_media = flow.data["driver_license"][0]
...     media_id, filename, decrypted_data = utils.flow_request_media_decryptor_sync(encrypted_media)
...     with open(filename, "wb") as file:
...         file.write(decrypted_data)
...     return types.FlowResponse(...)

Parameters:

encrypted_media (dict) – encrypted media data from the flow request (see example above).
dl_session (requests.Session | httpx.Client) – download session. Optional.

Returns:

tuple[str, str, bytes] - media_id (str): media ID - filename (str): media filename - decrypted_data (bytes): decrypted media file

Raises:

ValueError – If any of the hash verifications fail.

async pywa.utils.flow_request_media_decryptor_async(encrypted_media: dict[str, str | dict[str, str]], dl_session: AsyncClient | None = None) → tuple[str, str, bytes]#

Decrypt the encrypted media file from the flow request.

Read more at developers.facebook.com.
This implementation requires cryptography to be installed. To install it, run pip3 install 'pywa[cryptography]' or pip3 install cryptography.
This implementation is asynchronous. Use this if your flow request callback is asynchronous, otherwise use flow_request_media_decryptor_sync().

Example

>>> from pywa import WhatsApp, utils, types
>>> wa = WhatsApp(...)
>>> @wa.on_flow_request("/media-upload")
... async def on_media_upload_request(_: WhatsApp, flow: types.FlowRequest) -> types.FlowResponse | None:
...     encrypted_media = flow.data["driver_license"][0]
...     media_id, filename, decrypted_data = await utils.flow_request_media_decryptor_async(encrypted_media)
...     with open(filename, "wb") as file:
...         file.write(decrypted_data)
...     return types.FlowResponse(...)

Parameters:

encrypted_media (dict) – encrypted media data from the flow request (see example above).
dl_session (httpx.AsyncClient) – download session. Optional.

Returns:

tuple[str, str, bytes] - media_id (str): media ID - filename (str): media filename - decrypted_data (bytes): decrypted media file

Raises:

ValueError – If any of the hash verifications fail.

class pywa.handlers.FlowRequestCallbackWrapper(wa: WhatsApp, endpoint: str, callback: _FlowRequestHandlerT, acknowledge_errors: bool = True, handle_health_check: bool = True, private_key: str | None = None, private_key_password: str | None = None, request_decryptor: utils.FlowRequestDecryptor | None = None, response_encryptor: utils.FlowResponseEncryptor | None = None)#

This is a wrapper class for the flow request callback. It allows you to add more handlers to the same endpoint and split the logic into multiple functions.

THIS CLASS IS NOT MEANT TO BE INSTANTIATED DIRECTLY!

on(*, action: FlowRequestActionType, screen: str | None = None, data_filter: Callable[[WhatsApp, dict | None], bool] | None = None) → Callable[[_FlowRequestHandlerT], _FlowRequestHandlerT]#

Decorator to help you add more handlers to the same endpoint and split the logic into multiple functions.

Example

>>> wa = WhatsApp(...)
>>> @wa.on_flow_request("/feedback_flow")
>>> def feedback_flow_handler(_: WhatsApp, req: FlowRequest):
...    if req.has_error: print(req.data)
>>> @feedback_flow_handler.on(action=FlowRequestActionType.INIT)
>>> def on_init(_: WhatsApp, req: FlowRequest):
...     ...
>>> @feedback_flow_handler.on(action=FlowRequestActionType.DATA_EXCHANGE, screen="SURVEY")
>>> def on_survey_data_exchange(_: WhatsApp, req: FlowRequest):
...     ...
>>> @feedback_flow_handler.on(..., data_filter=lambda _, data: data.get("rating") == 5)
>>> def on_rating_5(_: WhatsApp, req: FlowRequest):
...     ...

Parameters:

action – The action type to listen to.
screen – The screen ID to listen to.
data_filter – A filter function to apply to the incoming data.

Returns:

The function itself.

on_errors() → Callable[[_FlowRequestHandlerT], _FlowRequestHandlerT]#

Decorator to add an error handler to the current endpoint.

Shortcut for set_errors_handler().

Example

>>> wa = WhatsApp(...)
>>> @wa.on_flow_request("/feedback_flow")
>>> def feedback_flow_handler(_: WhatsApp, req: FlowRequest):
...    ...
>>> @feedback_flow_handler.on_errors()
>>> def on_error(_: WhatsApp, req: FlowRequest):
...     logging.error("An error occurred: %s", req.data)

Returns:: The function itself.

add_handler(*, callback: _FlowRequestHandlerT, action: FlowRequestActionType, screen: str | None = None, data_filter: Callable[[WhatsApp, dict | None], bool] | None = None) → FlowRequestCallbackWrapper#

Add a handler to the current endpoint.

You can add multiple handlers to the same endpoint and split the logic into multiple functions.

Example

>>> wa = WhatsApp(...)
>>> def feedback_flow_handler(_: WhatsApp, req: FlowRequest):
...    ...
>>> on_init = lambda _, req: ...
>>> on_survey_data_exchange = lambda _, req: ...
>>> wa.add_flow_request_handler(
...     FlowRequestHandler(callback=feedback_flow_handler, endpoint="/feedback_flow")
... ).add_handler(callback=on_init, action=FlowRequestActionType.INIT)
...  .add_handler(callback=on_survey_data_exchange, action=FlowRequestActionType.DATA_EXCHANGE, screen="SURVEY")

Parameters:

callback – The callback function to handle this particular request.
action – The action type to listen to.
screen – The screen ID to listen to.
data_filter – A filter function to apply to the incoming data.

Returns:

The current instance.

set_errors_handler(callback: _FlowRequestHandlerT) → FlowRequestCallbackWrapper#

Add an error handler to the current endpoint.

Example

>>> wa = WhatsApp(...)
>>> def feedback_flow_handler(_: WhatsApp, req: FlowRequest):
...    ...
>>> def on_error(_: WhatsApp, req: FlowRequest):
...     logging.error("An error occurred: %s", req.data)
>>> wa.add_flow_request_handler(
...     FlowRequestHandler(callback=feedback_flow_handler, endpoint="/feedback_flow")
... ).set_errors_handler(callback=on_error)
...

Parameters:: callback – The callback function to handle errors.
Returns:: The current instance.
Raises:: ValueError – If an error handler is already set for this endpoint.

Flow Types

Contents

Flow Types#