Implementations of threaded and async versions of 'get_node' API with tests

container/save-and-restore.yml

@@ -1,12 +1,3 @@
-# Running elasticsearch in docker
-#   sudo docker compose -f save-and-restore.yml up -d
-# Test:
-#   curl -X GET "http://localhost:9200/"
-
-# .env file:
-#
-# HOST_EXTERNAL_IP_ADDRESS=192.168.50.49
-
 services:
   saveandrestore:
     image: ghcr.io/controlsystemstudio/phoebus/service-save-and-restore:master

container/start-save-and-restore.sh

@@ -3,6 +3,3 @@ set -x
 python create_env_file.py
 sudo docker compose -f save-and-restore.yml up -d
 python wait_for_startup.py
-
-# Wait until the service is started.
-#sleep 30

pyproject.toml

@@ -38,6 +38,7 @@ Homepage = "https://github.com/dmgav/save-and-restore-api"
 dev = [
   "pytest >=6",
   "pytest-cov >=3",
+  "pytest-asyncio",
   "pre-commit",
   "ruff",
 ]

src/save_and_restore_api/__init__.py

@@ -1,11 +1,6 @@
-"""
-Copyright (c) 2025 My Name. All rights reserved.
-
-save-and-restore-api: Python package for communication with CS Studio save-and-restore service
-"""
-
 from __future__ import annotations
 
+from ._api_threads import _SaveRestoreAPI_Threads as SaveRestoreAPI
 from ._version import version as __version__
 
-__all__ = ["__version__"]
+__all__ = ["__version__", "SaveRestoreAPI"]

src/save_and_restore_api/_api_async.py

@@ -0,0 +1,41 @@
+import httpx
+
+from ._api_base import _SaveRestoreAPI_Base
+
+
+class _SaveRestoreAPI_Async(_SaveRestoreAPI_Base):
+    def open(self):
+        self._client = httpx.AsyncClient(base_url=self._base_url, timeout=self._timeout)
+
+    async def close(self):
+        await self._client.aclose()
+        self._client = None
+
+    async def send_request(
+        self, method, url, *, params=None, url_params=None, headers=None, data=None, timeout=None, auth=None
+    ):
+        try:
+            client_response = None
+            kwargs = self._prepare_request(
+                method=method,
+                params=params,
+                url_params=url_params,
+                headers=headers,
+                data=data,
+                timeout=timeout,
+                auth=auth,
+            )
+            client_response = await self._client.request(method, url, **kwargs)
+            response = self._process_response(client_response=client_response)
+        except Exception:
+            response = self._process_comm_exception(method=method, params=params, client_response=client_response)
+
+        return response
+
+    async def login(self, *, username=None, password=None):
+        method, url, params = self._prepare_login(username=username, password=password)
+        await self.send_request(method, url, params=params)
+
+    async def get_node(self, node_uid):
+        method, url = self._prepare_get_node(node_uid=node_uid)
+        return await self.send_request(method, url)

src/save_and_restore_api/_api_base.py

@@ -0,0 +1,181 @@
+# import getpass
+import pprint
+from collections.abc import Mapping
+
+import httpx
+
+
+class RequestParameterError(Exception): ...
+
+
+class HTTPRequestError(httpx.RequestError): ...
+
+
+class HTTPClientError(httpx.HTTPStatusError): ...
+
+
+class HTTPServerError(httpx.HTTPStatusError): ...
+
+
+class RequestTimeoutError(TimeoutError):
+    def __init__(self, msg, request):
+        msg = f"Request timeout: {msg}"
+        self.request = request
+        super().__init__(msg)
+
+
+class RequestFailedError(Exception):
+    def __init__(self, request, response):
+        msg = response.get("msg", "") if isinstance(response, Mapping) else str(response)
+        msg = msg or "(no error message)"
+        msg = f"Request failed: {msg}"
+        self.request = request
+        self.response = response
+        super().__init__(msg)
+
+
+class _SaveRestoreAPI_Base:
+    RequestParameterError = RequestParameterError
+    RequestTimeoutError = RequestTimeoutError
+    RequestFailedError = RequestFailedError
+    HTTPRequestError = HTTPRequestError
+    HTTPClientError = HTTPClientError
+    HTTPServerError = HTTPServerError
+
+    def __init__(self, *, base_url, timeout, request_fail_exceptions=True):
+        self._base_url = base_url
+        self._timeout = timeout
+        self._client = None
+        self._root_node_uid = "44bef5de-e8e6-4014-af37-b8f6c8a939a2"
+        self._auth = None
+
+    @property
+    def ROOT_NODE_UID(self):
+        return self._root_node_uid
+
+    @staticmethod
+    def gen_auth(username, password):
+        return httpx.BasicAuth(username=username, password=password)
+
+    def set_auth(self, *, username, password):
+        self._auth = self.gen_auth(username=username, password=password)
+
+    # def set_username_password(self, username=None, password=None):
+    #     if not isinstance(username, str):
+    #         print("Username: ", end="")
+    #         username = input()
+    #     if not isinstance(password, str):
+    #         password = getpass.getpass()
+
+    #     self._username = username
+    #     self._password = password
+
+    # # TODO: rewrite the logic in this function
+    # def _check_response(self, *, request, response):
+    #     """
+    #     Check if response is a dictionary and has ``"success": True``. Raise an exception
+    #     if the request is considered failed and exceptions are allowed. If response is
+    #     a dictionary and contains no ``"success"``, then it is considered successful.
+    #     """
+    #     if self._request_fail_exceptions:
+    #         # The response must be a list or a dictionary. If the response is a dictionary
+    #         #   and the key 'success': False, then consider the request failed. If there
+    #         #   is not 'success' key, then consider the request successful.
+    #         is_iterable = isinstance(response, Iterable) and not isinstance(response, str)
+    #         is_mapping = isinstance(response, Mapping)
+    #         if not any([is_iterable, is_mapping]) or (is_mapping and not response.get("success", True)):
+    #             raise self.RequestFailedError(request, response)
+
+    def _process_response(self, *, client_response):
+        client_response.raise_for_status()
+        response = client_response.json()
+        return response
+
+    def _process_comm_exception(self, *, method, params, client_response):
+        """
+        The function must be called from ``except`` block and returns response with an error message
+        or raises an exception.
+        """
+        try:
+            raise
+
+        except httpx.TimeoutException as ex:
+            raise self.RequestTimeoutError(ex, {"method": method, "params": params}) from ex
+
+        except httpx.RequestError as ex:
+            raise self.HTTPRequestError(f"HTTP request error: {ex}") from ex
+
+        except httpx.HTTPStatusError as exc:
+            common_params = {"request": exc.request, "response": exc.response}
+            if client_response and (client_response.status_code < 500):
+                # Include more detail that httpx does by default.
+                message = (
+                    f"{exc.response.status_code}: "
+                    f"{exc.response.json()['detail'] if client_response.content else ''} "
+                    f"{exc.request.url}"
+                )
+                raise self.HTTPClientError(message, **common_params) from exc
+            else:
+                raise self.HTTPServerError(exc, **common_params) from exc
+
+    def _prepare_request(
+        self, *, method, params=None, url_params=None, headers=None, data=None, timeout=None, auth=None
+    ):
+        kwargs = {}
+        if params:
+            kwargs.update({"json": params})
+        if url_params:
+            kwargs.update({"params": url_params})
+        if headers:
+            kwargs.update({"headers": headers})
+        if data:
+            kwargs.update({"data": data})
+        if timeout is not None:
+            kwargs.update({"timeout": self._adjust_timeout(timeout)})
+        if method != "GET":
+            auth = auth or self._auth
+            if auth is not None:
+                kwargs.update({"auth": auth})
+        return kwargs
+
+    def _prepare_login(self, *, username=None, password=None):
+        method, url = "POST", "/login"
+        params = {"username": username, "password": password}
+        return method, url, params
+
+    def _prepare_get_node(self, *, node_uid):
+        method, url = "GET", f"/node/{node_uid}"
+        return method, url
+
+    def get_children(self, node_uid):
+        return self.send_request("GET", f"/node/{node_uid}/children")
+
+    def create_config(self, parent_node_uid, name, pv_list):
+        config_dict = {
+            "configurationNode": {
+                "name": name,
+                "nodeType": "CONFIGURATION",
+                "userName": self._username,
+            },
+            "configurationData": {
+                "pvList": pv_list,
+            },
+        }
+        print(f"config_dict=\n{pprint.pformat(config_dict)}")
+        return self.send_request("PUT", f"/config?parentNodeId={parent_node_uid}", json=config_dict)
+
+    def update_config(self, node_uid, name, pv_list):
+        config_dict = {
+            "configurationNode": {
+                "name": name,
+                "nodeType": "CONFIGURATION",
+                "userName": self._username,
+                "uniqueId": node_uid,
+            },
+            "configurationData": {
+                "pvList": pv_list,
+            },
+        }
+        print(f"config_dict=\n{pprint.pformat(config_dict)}")
+        # return self.send_request("POST", f"/config/{node_uid}", json=config_dict)
+        return self.send_request("POST", "/config", json=config_dict)

src/save_and_restore_api/_api_threads.py

@@ -0,0 +1,41 @@
+import httpx
+
+from ._api_base import _SaveRestoreAPI_Base
+
+
+class _SaveRestoreAPI_Threads(_SaveRestoreAPI_Base):
+    def open(self):
+        self._client = httpx.Client(base_url=self._base_url, timeout=self._timeout)
+
+    def close(self):
+        self._client.close()
+        self._client = None
+
+    def send_request(
+        self, method, url, *, params=None, url_params=None, headers=None, data=None, timeout=None, auth=None
+    ):
+        try:
+            client_response = None
+            kwargs = self._prepare_request(
+                method=method,
+                params=params,
+                url_params=url_params,
+                headers=headers,
+                data=data,
+                timeout=timeout,
+                auth=auth,
+            )
+            client_response = self._client.request(method, url, **kwargs)
+            response = self._process_response(client_response=client_response)
+        except Exception:
+            response = self._process_comm_exception(method=method, params=params, client_response=client_response)
+
+        return response
+
+    def login(self, *, username=None, password=None):
+        method, url, params = self._prepare_login(username=username, password=password)
+        self.send_request(method, url, params=params)
+
+    def get_node(self, node_uid):
+        method, url = self._prepare_get_node(node_uid=node_uid)
+        return self.send_request(method, url)

src/save_and_restore_api/aio/__init__.py

@@ -0,0 +1,6 @@
+from __future__ import annotations
+
+from .._api_async import _SaveRestoreAPI_Async as SaveRestoreAPI
+from .._version import version as __version__
+
+__all__ = ["__version__", "SaveRestoreAPI"]