feat: action versioning implementation (#1196)

Co-authored-by: angrybayblade <[email protected]>

python/.gitignore

@@ -0,0 +1 @@
+*.composio.lock

python/composio/client/collections.py

@@ -1027,16 +1027,19 @@ class ActionModel(BaseModel):
     """Action data model."""
 
     name: str
-    display_name: t.Optional[str] = None
+    description: str
     parameters: ActionParametersModel
     response: ActionResponseModel
     appName: str
     appId: str
-    tags: t.List[str]
-    enabled: bool = False
+    version: str
+    available_versions: t.List[str]
 
+    tags: t.List[str]
     logo: t.Optional[str] = None
-    description: t.Optional[str] = None
+
+    display_name: t.Optional[str] = None
+    enabled: bool = False
 
 
 ParamPlacement = t.Literal["header", "path", "query", "subdomain", "metadata"]
@@ -1080,8 +1083,19 @@ class Actions(Collection[ActionModel]):
     model = ActionModel
     endpoint = v2.actions
 
-    # TODO: Overload
-    def get(  # type: ignore
+    def _get_action(self, action: ActionType) -> ActionModel:
+        return self.model(
+            **self._raise_if_required(
+                response=self.client.http.get(
+                    url=str(self.endpoint / str(action)),
+                    params={
+                        "version": Action(action).version,
+                    },
+                )
+            ).json()
+        )
+
+    def _get_actions(
         self,
         actions: t.Optional[t.Sequence[ActionType]] = None,
         apps: t.Optional[t.Sequence[AppType]] = None,
@@ -1090,18 +1104,6 @@ def get(  # type: ignore
         use_case: t.Optional[str] = None,
         allow_all: bool = False,
     ) -> t.List[ActionModel]:
-        """
-        Get a list of apps by the specified filters.
-
-        :param actions: Filter by the list of Actions.
-        :param apps: Filter by the list of Apps.
-        :param tags: Filter by the list of given Tags.
-        :param limit: Limit the number of actions to a specific number.
-        :param use_case: Filter by use case.
-        :param allow_all: Allow querying all of the actions for a specific
-                        app
-        :return: List of actions
-        """
 
         def is_action(obj):
             try:
@@ -1224,6 +1226,76 @@ def is_action(obj):
             items = [self.model(**item) for item in local_items] + items
         return items
 
+    @t.overload  # type: ignore
+    def get(self) -> t.List[ActionModel]: ...
+
+    @t.overload  # type: ignore
+    def get(self, action: t.Optional[ActionType] = None) -> ActionModel: ...
+
+    @t.overload  # type: ignore
+    def get(
+        self,
+        actions: t.Optional[t.Sequence[ActionType]] = None,
+        apps: t.Optional[t.Sequence[AppType]] = None,
+        tags: t.Optional[t.Sequence[TagType]] = None,
+        limit: t.Optional[int] = None,
+        use_case: t.Optional[str] = None,
+        allow_all: bool = False,
+    ) -> t.List[ActionModel]: ...
+
+    def get(  # type: ignore
+        self,
+        action: t.Optional[ActionType] = None,
+        *,
+        actions: t.Optional[t.Sequence[ActionType]] = None,
+        apps: t.Optional[t.Sequence[AppType]] = None,
+        tags: t.Optional[t.Sequence[TagType]] = None,
+        limit: t.Optional[int] = None,
+        use_case: t.Optional[str] = None,
+        allow_all: bool = False,
+    ) -> t.Union[ActionModel, t.List[ActionModel]]:
+        """
+        Get a list of apps by the specified filters.
+
+        :param actions: Filter by the list of Actions.
+        :param action: Get data for this action.
+        :param apps: Filter by the list of Apps.
+        :param tags: Filter by the list of given Tags.
+        :param limit: Limit the number of actions to a specific number.
+        :param use_case: Filter by use case.
+        :param allow_all: Allow querying all of the actions for a specific
+                        app
+        :return: List of actions
+        """
+        if action is not None:
+            return self._get_action(action=action)
+
+        return self._get_actions(
+            actions=actions,
+            apps=apps,
+            tags=tags,
+            limit=limit,
+            use_case=use_case,
+            allow_all=allow_all,
+        )
+
+    @staticmethod
+    def _serialize_auth(auth: t.Optional[CustomAuthObject]) -> t.Optional[t.Dict]:
+        if auth is None:
+            return None
+
+        data = auth.model_dump(exclude_none=True)
+        data["parameters"] = [
+            {"in": d["in_"], "name": d["name"], "value": d["value"]}
+            for d in data["parameters"]
+        ]
+        for param in data["parameters"]:
+            if param["in"] == "metadata":
+                raise ComposioClientError(
+                    f"Param placement cannot be 'metadata' for remote action execution: {param}"
+                )
+        return data
+
     def execute(
         self,
         action: Action,
@@ -1297,6 +1369,7 @@ def execute(
                         "appName": action.app,
                         "input": modified_params,
                         "text": text,
+                        "version": action.version,
                         "sessionInfo": {
                             "sessionId": session_id,
                         },
@@ -1319,6 +1392,7 @@ def execute(
                     "appName": action.app,
                     "input": modified_params,
                     "text": text,
+                    "version": action.version,
                     "authConfig": self._serialize_auth(auth=auth),
                     "sessionInfo": {
                         "sessionId": session_id,
@@ -1353,23 +1427,6 @@ def request(
             },
         ).json()
 
-    @staticmethod
-    def _serialize_auth(auth: t.Optional[CustomAuthObject]) -> t.Optional[t.Dict]:
-        if auth is None:
-            return None
-
-        data = auth.model_dump(exclude_none=True)
-        data["parameters"] = [
-            {"in": d["in_"], "name": d["name"], "value": d["value"]}
-            for d in data["parameters"]
-        ]
-        for param in data["parameters"]:
-            if param["in"] == "metadata":
-                raise ComposioClientError(
-                    f"Param placement cannot be 'metadata' for remote action execution: {param}"
-                )
-        return data
-
     def search_for_a_task(
         self,
         use_case: str,

python/composio/client/enums/action.py

@@ -3,17 +3,37 @@
 
 from composio.client.enums.base import ActionData, replacement_action_name
 from composio.client.enums.enum import Enum, EnumGenerator
+from composio.constants import VERSION_LATEST, VERSION_LATEST_BASE
 from composio.exceptions import ComposioSDKError
 
 
 _ACTION_CACHE: t.Dict[str, "Action"] = {}
 
 
+class InvalidVersionString(ComposioSDKError):
+
+    def __init__(self, message: str, *args: t.Any, delegate: bool = False) -> None:
+        super().__init__(message, *args, delegate=delegate)
+
+
+def clean_version_string(version: str) -> str:
+    version = version.lower()
+    if version in (VERSION_LATEST, VERSION_LATEST_BASE):
+        return version
+
+    version = "_".join(version.split(".")).lstrip("v")
+    if version.count("_") != 1:
+        raise InvalidVersionString(version)
+    return version
+
+
 class Action(Enum[ActionData], metaclass=EnumGenerator):
     cache_folder = "actions"
     cache = _ACTION_CACHE
     storage = ActionData
 
+    _version: t.Optional[str] = None
+
     def load(self) -> ActionData:
         """Handle deprecated actions"""
         action_data = super().load()
@@ -121,3 +141,38 @@ def is_local(self) -> bool:
     def is_runtime(self) -> bool:
         """If set `True` the `app` is a runtime app."""
         return self.load().is_runtime
+
+    @property
+    def is_version_set(self) -> bool:
+        """If `True` version is set explicitly."""
+        return self._version is not None
+
+    @property
+    def version(self) -> str:
+        """Version string for the action enum instance."""
+        return self._version or self.load().version
+
+    @property
+    def available_versions(self) -> t.List[str]:
+        """List of available version strings."""
+        return self.load().available_version
+
+    def with_version(self, version: str) -> "Action":
+        if self.is_local:
+            raise RuntimeError("Versioning is not allowed for local tools")
+
+        action = Action(self.slug, cache=False)
+        action._data = self.load()  # pylint: disable=protected-access
+        action._version = clean_version_string(  # pylint: disable=protected-access
+            version=version
+        )
+        return action
+
+    def latest(self) -> "Action":
+        return self.with_version(version=VERSION_LATEST)
+
+    def latest_base(self) -> "Action":
+        return self.with_version(version=VERSION_LATEST_BASE)
+
+    def __matmul__(self, other: str) -> "Action":
+        return self.with_version(version=other)

python/composio/client/enums/action.pyi

@@ -7,11 +7,15 @@ from composio.exceptions import ComposioSDKError
 
 _ACTION_CACHE: t.Dict[str, "Action"] = {}
 
+def clean_version_string(version: str) -> str: ...
+
 class Action(Enum[ActionData], metaclass=EnumGenerator):
     cache_folder = "actions"
     cache = _ACTION_CACHE
     storage = ActionData
 
+    _version: t.Optional[str] = None
+
     def load(self) -> ActionData: ...
     def load_from_runtime(self) -> t.Optional[ActionData]: ...
     def fetch_and_cache(self) -> t.Optional[ActionData]: ...
@@ -27,6 +31,17 @@ class Action(Enum[ActionData], metaclass=EnumGenerator):
     def is_local(self) -> bool: ...
     @property
     def is_runtime(self) -> bool: ...
+    @property
+    def is_version_set(self) -> bool: ...
+    @property
+    def version(self) -> str: ...
+    @property
+    def available_versions(self) -> t.List[str]: ...
+    def with_version(self, version: str) -> "Action": ...
+    def latest(self) -> "Action": ...
+    def latest_base(self) -> "Action": ...
+    def __matmul__(self, other: str) -> "Action": ...
+
     AFFINITY_GET_ALL_COMPANIES: "Action"
     AFFINITY_GET_ALL_LIST_ENTRIES_ON_A_LIST: "Action"
     AFFINITY_GET_ALL_LIST_ENTRIES_ON_A_SAVED_VIEW: "Action"

python/composio/client/enums/base.py

@@ -5,7 +5,14 @@
 import difflib
 import typing as t
 
-from composio.constants import LOCAL_CACHE_DIRECTORY
+from pydantic import Field
+
+from composio.constants import (
+    COMPOSIO_VERSIONING_POLICY,
+    LOCAL_CACHE_DIRECTORY,
+    VERSION_LATEST,
+    VERSION_LATEST_BASE,
+)
 from composio.exceptions import ComposioSDKError
 from composio.storage.base import LocalStorage
 
@@ -86,6 +93,17 @@ class ActionData(LocalStorage):
     replaced_by: t.Optional[str] = None
     "If set, the action is deprecated and replaced by the given action."
 
+    version: str = COMPOSIO_VERSIONING_POLICY
+    "Specify what version to use when executing action."
+
+    available_version: t.List[str] = Field(
+        default_factory=lambda: [
+            VERSION_LATEST,
+            VERSION_LATEST_BASE,
+        ]
+    )
+    "Specify what version to use when executing action."
+
 
 class TriggerData(LocalStorage):
     """Local storage for `Trigger` object."""

python/composio/client/enums/enum.py

@@ -24,7 +24,11 @@ class Enum(t.Generic[DataT]):
     cache: t.Dict[str, "te.Self"]
     storage: t.Type[DataT]
 
-    def __new__(cls, value: t.Union[str, te.Self, t.Type[SentinalObject]]) -> "te.Self":
+    def __new__(
+        cls,
+        value: t.Union[str, te.Self, t.Type[SentinalObject]],
+        cache: bool = True,
+    ) -> "te.Self":
         """Cache the enum singleton."""
         # No caching for runtime actions
         if hasattr(value, "sentinel"):  # TODO: get rid of SentinalObject
@@ -41,14 +45,19 @@ def __new__(cls, value: t.Union[str, te.Self, t.Type[SentinalObject]]) -> "te.Se
         value = value.upper()
 
         cached_enum = cls.cache.get(value)
-        if cached_enum is not None:
+        if cache and cached_enum is not None:
             return cached_enum  # type: ignore[return-value]
 
         enum = super().__new__(cls)
-        cls.cache[value] = enum
+        if cache:
+            cls.cache[value] = enum
         return enum
 
-    def __init__(self, value: t.Union[str, te.Self, t.Type[SentinalObject]]) -> None:
+    def __init__(
+        self,
+        value: t.Union[str, te.Self, t.Type[SentinalObject]],
+        cache: bool = True,  # pylint: disable=unused-argument
+    ) -> None:
         if hasattr(self, "_data"):
             # Object was pulled from cache and is already initialized
             return