[utils] adding Version to provide the latest versions to the api & …

…flows, and to perform min checks

pywa/client.py

@@ -54,7 +54,10 @@ def __init__(
         phone_id: str | int,
         token: str,
         base_url: str = "https://graph.facebook.com",
-        api_version: float | int = 18.0,
+        api_version: str
+        | int
+        | float
+        | Literal[utils.Version.GRAPH_API] = utils.Version.GRAPH_API,
         session: requests.Session | None = None,
         server: Flask | FastAPI | None = None,
         webhook_endpoint: str = "/",
@@ -105,7 +108,7 @@ def __init__(
             token: The token of the WhatsApp business account (In production, you should
              `use permanent token <https://developers.facebook.com/docs/whatsapp/business-management-api/get-started>`_).
             base_url: The base URL of the WhatsApp API (Do not change unless you know what you're doing).
-            api_version: The API version of the WhatsApp API (default: ``18.0``).
+            api_version: The API version of the WhatsApp Cloud API (default to the latest version).
             session: The session to use for requests (default: new ``requests.Session()``, For cases where you want to
              use a custom session, e.g. for proxy support. Do not use the same session across multiple WhatsApp clients!).
             server: The Flask or FastAPI app instance to use for the webhook. required when you want to handle incoming
@@ -139,6 +142,16 @@ def __init__(
         if not phone_id or not token:
             raise ValueError("phone_id and token must be provided.")
 
+        try:
+            utils.Version.GRAPH_API.validate_min_version(str(api_version))
+        except ValueError:
+            warnings.warn(
+                message=f"PyWa officially supports WhatsApp Cloud API version {utils.Version.GRAPH_API.min} and up. "
+                f"Using version {api_version} is not officially supported and may cause unexpected behavior.",
+                category=RuntimeWarning,
+                stacklevel=2,
+            )
+
         self._phone_id = str(phone_id)
         self.filter_updates = filter_updates
         self.business_account_id = (
@@ -149,7 +162,7 @@ def __init__(
             token=token,
             session=session or requests.Session(),
             base_url=base_url,
-            api_version=float(api_version),
+            api_version=float(str(api_version)),
         )
 
         self._handlers: dict[
@@ -315,7 +328,6 @@ def send_message(
             ...         title="Feedback",
             ...         flow_id="1234567890",
             ...         flow_token="AQAAAAACS5FpgQ_cAAAAAD0QI3s.",
-            ...         flow_message_version="3",
             ...         flow_action_type=FlowActionType.NAVIGATE,
             ...         flow_action_screen="RECOMMENDED"
             ...     ),
@@ -1546,8 +1558,8 @@ def update_flow_metadata(
             flow_id: The flow ID.
             name: The name of the flow (optional).
             categories: The new categories of the flow (optional).
-            endpoint_uri: The URL of the WA FlowJSON Endpoint. Starting from FlowJSON JSON version 3.0 this property should be
-             specified only via API. Do not provide this field if you are cloning a FlowJSON with FlowJSON JSON version below 3.0.
+            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.
 
         Example:
 
@@ -1556,7 +1568,8 @@ def update_flow_metadata(
             >>> wa.update_flow_metadata(
             ...     flow_id='1234567890',
             ...     name='Feedback',
-            ...     categories=[FlowCategory.SURVEY, FlowCategory.OTHER]
+            ...     categories=[FlowCategory.SURVEY, FlowCategory.OTHER],
+            ...     endpoint_uri='https://my-api-server/feedback_flow'
             ... )
 
         Returns:

pywa/types/callback.py

@@ -33,6 +33,7 @@
 from .base_update import BaseUserUpdate  # noqa
 from .flows import FlowStatus, FlowActionType
 from .others import MessageType, Metadata, ReplyToMessage, User
+from .. import utils
 
 if TYPE_CHECKING:
     from pywa.client import WhatsApp
@@ -577,8 +578,8 @@ class FlowButton:
     Attributes:
         title: Text on the CTA button. e.g ``SignUp``, Up to 20 characters, no emojis)
         flow_id: Unique ID of the Flow provided by WhatsApp.
-        flow_token: Flow token that is generated by the business to serve as an identifier for data exchange.
-        flow_message_version: Version of the flow message. Currently, the value must be 3.
+        flow_token: Flow token generated by the business to serve as an identifier for data exchange.
+        flow_message_version: Version of the flow message. Default is the latest version.
         flow_action_type: Type of action to be performed when the user clicks on the CTA button.
         flow_action_screen: The ID of the first Screen.
          Required when ``flow_action_type`` is ``FlowActionType.NAVIGATE`` (default).
@@ -592,20 +593,25 @@ class FlowButton:
     flow_token: str
     flow_action_type: Literal[
         FlowActionType.NAVIGATE, FlowActionType.DATA_EXCHANGE
-    ] | str | None = None
+    ] | None = None
     flow_action_screen: str | None = None
     flow_action_payload: dict[str, Any] | None = None
-    flow_message_version: int | str = 3
+    flow_message_version: int | float | str | Literal[
+        utils.Version.FLOW_MSG
+    ] = utils.Version.FLOW_MSG
     mode: Literal[FlowStatus.PUBLISHED, FlowStatus.DRAFT] = FlowStatus.PUBLISHED
 
-    def to_dict(self) -> dict:
+    def __post_init__(self):
+        utils.Version.FLOW_MSG.validate_min_version(str(self.flow_message_version))
         if (
             self.flow_action_type == FlowActionType.NAVIGATE
             and self.flow_action_screen is None
         ):
             raise ValueError(
                 "flow_action_screen cannot be None when flow_action_type is FlowActionType.NAVIGATE"
             )
+
+    def to_dict(self) -> dict:
         return {
             "name": "flow",
             "parameters": {

pywa/types/flows.py

@@ -7,7 +7,7 @@
 import datetime
 import json
 import pathlib
-from typing import Iterable, TYPE_CHECKING, Any, BinaryIO
+from typing import Iterable, TYPE_CHECKING, Any, BinaryIO, Literal
 
 from pywa import utils
 from pywa.types.base_update import BaseUserUpdate  # noqa
@@ -548,18 +548,31 @@ class FlowJSON:
     - Read more at `developers.facebook.com <https://developers.facebook.com/docs/whatsapp/flows/reference/flowjson>`_.
 
     Attributes:
-        version: The Flow JSON version (Read more at `developers.facebook.com <https://developers.facebook.com/docs/whatsapp/flows/reference/versioning>`_).
         screens: The screens that are part of the flow (Read more at `developers.facebook.com <https://developers.facebook.com/docs/whatsapp/flows/reference/flowjson#screens>`_).
-        data_api_version: The version to use during communication with the WhatsApp Flows Data Endpoint. Required if the data channel is set.
-        data_channel_uri: The endpoint to use to communicate with your server (If you using the WhatsApp Flows Data Endpoint).
+        version: The Flow JSON version. Default to latest (Read more at `developers.facebook.com <https://developers.facebook.com/docs/whatsapp/flows/reference/versioning>`_).
+        data_api_version: The version to use during communication with the WhatsApp Flows Data Endpoint. Default to latest. Required if the data channel is set.
         routing_model: Defines the rules for the screen by limiting the possible state transition. (Read more at `developers.facebook.com <https://developers.facebook.com/docs/whatsapp/flows/reference/flowjson#routing-model>`_).
+        data_channel_uri: The endpoint to use to communicate with your server (When using v3.0 or higher, this field need to be set via :meth:`WhatsApp.update_flow_metadata`).
     """
 
-    version: str
     screens: Iterable[Screen]
-    data_api_version: str | None = None
+    version: str | float | Literal[utils.Version.FLOW_JSON] = utils.Version.FLOW_JSON
+    data_api_version: str | float | Literal[utils.Version.FLOW_DATA_API] | None = None
+    routing_model: dict[str, Iterable[str]] | None = None
     data_channel_uri: str | None = None
-    routing_model: dict[str, Iterable[str]] = dataclasses.field(default_factory=dict)
+
+    def __post_init__(self):
+        self.version = str(self.version)
+        utils.Version.FLOW_JSON.validate_min_version(self.version)
+        if self.data_channel_uri and float(self.version) >= 3.0:
+            raise ValueError(
+                "When using v3.0 or higher, `data_channel_uri` need to be set via WhatsApp.update_flow_metadata.\n"
+                ">>> wa = WhatsApp(...)\n"
+                f">>> wa.update_flow_metadata(flow_id, endpoint_uri={self.data_channel_uri!r})\n"
+            )
+        if self.data_api_version:
+            self.data_api_version = str(self.data_api_version)
+            utils.Version.FLOW_DATA_API.validate_min_version(self.data_api_version)
 
     def to_dict(self):
         return dataclasses.asdict(
@@ -598,7 +611,12 @@ class Screen:
 
 
 class LayoutType(utils.StrEnum):
-    """LayoutType is the type of layout that is used to display the components."""
+    """
+    The type of layout that is used to display the components.
+
+    Attributes:
+        SINGLE_COLUMN: A single column layout.
+    """
 
     SINGLE_COLUMN = "SingleColumnLayout"
 
@@ -1248,7 +1266,7 @@ class FlowActionType(utils.StrEnum):
     Attributes:
         COMPLETE: Triggers the termination of the Flow with the provided payload
          (Read more at `developers.facebook.com <https://developers.facebook.com/docs/whatsapp/flows/reference/flowjson#complete-action>`_).
-        DATA_EXCHANGE: TSending Data to WhatsApp Flows Data Endpoint
+        DATA_EXCHANGE: Sending Data to WhatsApp Flows Data Endpoint
          (Read more at `developers.facebook.com <https://developers.facebook.com/docs/whatsapp/flows/reference/flowjson#data-exchange-action>`_).
         NAVIGATE: Triggers the next screen with the payload as its input. The CTA button will be disabled until the
          payload with data required for the next screen is supplied.
@@ -1304,3 +1322,13 @@ class Action:
     name: FlowActionType | str
     next: ActionNext | None = None
     payload: dict[str, str | DataKey | FormRef] | None = None
+
+    def __post_init__(self):
+        if self.name == FlowActionType.NAVIGATE.value:
+            if self.next is None:
+                raise ValueError("next is required for FlowActionType.NAVIGATE")
+        if self.name == FlowActionType.COMPLETE.value:
+            if self.payload is None:
+                raise ValueError(
+                    "payload is required for FlowActionType.COMPLETE (use {} for empty payload)"
+                )

pywa/types/template.py

@@ -883,9 +883,7 @@ class FlowButton(NewButtonABC):
         )
         title: str
         flow_id: str | int
-        flow_action: Literal[
-            FlowActionType.NAVIGATE, FlowActionType.DATA_EXCHANGE
-        ] | str
+        flow_action: Literal[FlowActionType.NAVIGATE, FlowActionType.DATA_EXCHANGE]
         navigate_screen: str | None = None
 
         def __post_init__(self):

pywa/utils.py

@@ -33,6 +33,44 @@ def is_cryptography_installed():
         return False
 
 
+class Version(enum.Enum):
+    """
+    Enum for the latest and minimum versions of the WhatsApp API.
+
+    - Use the constant to get the latest version. Example: ``WhatsApp(..., api_version=Version.GRAPH_API)``
+    - Use the ``min`` attribute to get the minimum version. Example: Version.GRAPH_API.min
+
+    Attributes:
+        GRAPH_API: (MIN_VERSION: str, LATEST_VERSION: str)
+        FLOW_JSON: (MIN_VERSION: str, LATEST_VERSION: str)
+        FLOW_DATA_API: (MIN_VERSION: str, LATEST_VERSION: str)
+        FLOW_MSG: (MIN_VERSION: str, LATEST_VERSION: str)
+    """
+
+    # KEY = (MIN_VERSION: str, LATEST_VERSION: str)
+    GRAPH_API = ("17.0", "18.0")
+    FLOW_JSON = ("2.1", "3.0")
+    FLOW_DATA_API = ("3.0", "3.0")
+    FLOW_MSG = ("3", "3")
+
+    def __new__(cls, min_version: str, latest_version: str):
+        obj = object.__new__(cls)
+        obj._value_ = latest_version
+        obj.min = min_version
+        return obj
+
+    def __str__(self):
+        """Required for the ``Version`` enum to be used as a string."""
+        return self.value
+
+    def validate_min_version(self, version: str):
+        """Check if the given version is supported."""
+        if float(version) < float(self.min):
+            raise ValueError(
+                f"{self.name}: version {version} is not supported. Minimum version is {self.min}."
+            )
+
+
 class StrEnum(str, enum.Enum):
     """Enum where the values are also (and must be) strings."""
 

tests/data/flows/2.1/examples.json

@@ -1,7 +1,6 @@
 {
   "customer_satisfaction_survey": {
     "version": "2.1",
-    "routing_model": {},
     "screens": [
       {
         "id": "RECOMMEND",
@@ -196,7 +195,6 @@
   },
   "load_re_engagement": {
     "version": "2.1",
-    "routing_model": {},
     "screens": [
       {
         "id": "SIGN_UP",
@@ -256,7 +254,6 @@
   },
   "costumer_engagement": {
     "version": "2.1",
-    "routing_model": {},
     "screens": [
       {
         "id": "QUESTION_ONE",
@@ -456,7 +453,6 @@
   },
   "support_request": {
     "version": "2.1",
-    "routing_model": {},
     "screens": [
       {
         "id": "DETAILS",
@@ -541,7 +537,6 @@
   },
   "communication_preferences": {
     "version": "2.1",
-    "routing_model": {},
     "screens": [
       {
         "id": "PREFERENCES",
@@ -619,7 +614,6 @@
   },
   "register_for_an_event": {
     "version": "2.1",
-    "routing_model": {},
     "screens": [
       {
         "id": "SIGN_UP",

tests/test_flows.py

@@ -1,5 +1,7 @@
 import json
 
+import pytest
+
 from pywa.types.flows import (
     FlowJSON,
     Screen,
@@ -1150,4 +1152,27 @@ def test_flows_to_json():
     with open(f"tests/data/flows/{FLOWS_VERSION}/examples.json", "r") as f:
         examples = json.load(f)
     for flow_name, flow in FLOWS.items():
-        assert flow.to_dict() == examples[flow_name]
+        try:
+            assert flow.to_dict() == examples[flow_name]
+        except AssertionError:
+            raise AssertionError(
+                f"Flow {flow_name} does not match example\nFlow: {flow}\nJSON: {examples[flow_name]}"
+            )
+
+
+def test_min_version():
+    with pytest.raises(ValueError):
+        FlowJSON(version="1.0", screens=[])
+
+
+def test_data_channel_uri():
+    with pytest.raises(ValueError):
+        FlowJSON(version="3.0", data_channel_uri="https://example.com", screens=[])
+
+
+def test_action():
+    with pytest.raises(ValueError):
+        Action(name=FlowActionType.NAVIGATE)
+
+    with pytest.raises(ValueError):
+        Action(name=FlowActionType.COMPLETE)