feat: Add AsApi.get_actions()

Author: a.b.christie

README.rst

@@ -119,6 +119,7 @@ The following Squonk2 Account Server API functions are available: -
 - ``AsApi.disable_asset()``
 - ``AsApi.enable_asset()``
 - ``AsApi.get_asset()``
+- ``AsApi.get_actions()``
 - ``AsApi.get_available_assets()``
 - ``AsApi.get_available_units()``
 - ``AsApi.get_available_products()``

src/squonk2/api.py

@@ -12,7 +12,7 @@
 
 @dataclass
 class ApiRv:
-    """The return value from most of the the AsApi class public methods.
+    """The return value from most of the class public methods.
 
     :param success: True if the call was successful, False otherwise.
     :param msg: API request response content

src/squonk2/as_api.py

@@ -1750,3 +1750,55 @@ def get_default_organisation(
             error_message="Failed getting default organisation",
             timeout=timeout_s,
         )[0]
+
+    @classmethod
+    @synchronized
+    def get_actions(
+        cls,
+        access_token: str,
+        *,
+        from_: Optional[date] = None,
+        until: Optional[date] = None,
+        merchant_id: int = 0,
+        product_id: str = "",
+        unit_id: str = "",
+        org_id: str = "",
+        action_format: EventStreamFormat = EventStreamFormat.PROTOCOL_STRING,
+        timeout_s: int = _READ_TIMEOUT_S,
+    ) -> ApiRv:
+        """Gets the Action records (protocol buffers).
+
+        :param access_token: A valid AS API access token
+        :param action_format: The action format
+        :param from_: An optional start
+        :param until: An optional end
+        :param merchant_id: An optional merchant
+        :param product_id: An optional Product UUID
+        :param unit_id: An optional Unit UUID
+        :param org_id: An optional Organisation UUID
+        :param timeout_s: The underlying request timeout
+        """
+        assert access_token
+
+        params: Dict[str, Any] = {"format": action_format.name}
+        if from_:
+            params["from"] = str(from_)
+        if until:
+            params["until"] = str(until)
+        if merchant_id:
+            params["m_id"] = merchant_id
+        if product_id:
+            params["product_id"] = product_id
+        if unit_id:
+            params["unit_id"] = unit_id
+        if org_id:
+            params["org_id"] = org_id
+
+        return AsApi.__request(
+            "GET",
+            "/action",
+            access_token=access_token,
+            params=params,
+            error_message="Failed getting actions",
+            timeout=timeout_s,
+        )[0]