feat(openHEXA): add openHEXA client

docs/hexa.md

@@ -0,0 +1,43 @@
+
+# OpenHEXA Toolbox OpenHEXAClient
+
+The OpenHEXAClient module allow users to interact with the OpenHEXA backend using GraphQL syntax.
+
+* [Installation](#installation)
+* [Example](#example)
+
+
+## [Installation](#)
+
+``` sh
+pip install openhexa.toolbox
+```
+
+## [Example](#)
+
+Import OpenHEXA module:
+```python
+import json
+from openhexa.toolbox.hexa import OpenHEXA
+# We can authenticate using username / password
+hexa_client = OpenHEXA("https://app.demo.openhexa.org", username="username",  password="password")
+
+# You can also use the token provided by OpenHEXA on the pipelines 
+hexa_client = OpenHEXA("https://app.demo.openhexa.org", token="token")
+
+workspaces = hexa.query("""
+    query {
+        workspaces (page: 1, perPage: 10) {
+            items {
+                slug
+                name
+            }
+        }
+    }
+""")["workspaces"]["items"]
+print("Workspaces:")
+print(json.dumps(workspaces, indent=2))
+```
+
+
+

openhexa/toolbox/hexa/__init__.py

@@ -0,0 +1,4 @@
+from .api import OpenHEXAClient
+from .hexa import OpenHEXA
+
+__all__ = ["OpenHEXA", "OpenHEXAClient"]

openhexa/toolbox/hexa/api.py

@@ -0,0 +1,61 @@
+from requests import Session
+import typing
+
+
+class OpenHEXAClient:
+    def __init__(self, base_url):
+        self.url = base_url.rstrip("/")
+        self.session = Session()
+        self.session.headers.update({"Content-Type": "application/json", "User-Agent": "OpenHEXA Python Client"})
+
+    def authenticate(
+        self,
+        with_credentials: typing.Optional[tuple[str, str]] = None,
+        with_token: typing.Optional[str] = None,
+    ):
+        """
+        with_credentials: tuple of email and password
+        with_token: JWT token
+        """
+        if with_credentials:
+            resp = self._graphql_request(
+                """
+                mutation Login($input: LoginInput!) {
+                    login(input: $input) {
+                        success
+                    }
+                }
+            """,
+                {
+                    "input": {
+                        "email": with_credentials[0],
+                        "password": with_credentials[1],
+                    }
+                },
+            )
+            resp.raise_for_status()
+            data = resp.json()["data"]
+            if data["login"]["success"]:
+                self.session.headers["Cookie"] = resp.headers["Set-Cookie"]
+            else:
+                raise Exception("Login failed")
+        elif with_token:
+            self.session.headers.update({"Authorization": f"Bearer {with_token}"})
+        try:
+            self.query("""query{me {user {id}}}""")
+            return True
+        except Exception:
+            raise Exception("Authentication failed")
+
+    def _graphql_request(self, operation, variables=None):
+        return self.session.post(f"{self.url}/graphql", json={"query": operation, "variables": variables})
+
+    def query(self, operation, variables=None):
+        resp = self._graphql_request(operation, variables)
+        if resp.status_code == 400:
+            raise Exception(resp.json()["errors"][0]["message"])
+        resp.raise_for_status()
+        payload = resp.json()
+        if payload.get("errors"):
+            raise Exception(payload["errors"])
+        return payload["data"]

openhexa/toolbox/hexa/hexa.py

@@ -0,0 +1,85 @@
+from .api import OpenHEXAClient
+
+
+class OpenHEXA:
+    def __init__(self, server_url: str, username: str | None, password: str | None, token: str | None = None) -> None:
+        """
+        Initializes the openHEXA client. If username and password are provided we will try to
+        authenticate via credentials, else wi
+
+        Parameters
+        ----------
+        server_url: openHEXA server URL
+        username: openHEXA instance username
+        password: openHEXA instance password
+        token: openHEXA pipeline token
+
+        Raises
+        ------
+        Exception
+            when a login or authentication error
+
+        Examples:
+            >>> from openhexa.toolbox.hexa import openHEXA
+            >>> hexa = openHEXA(server_url="http://iaso-staging.bluesquare.org",
+            >>>             username="user",
+            >>>             password="pass")
+
+            or
+
+            >>> from openhexa.toolbox.hexa import openHEXA
+            >>> hexa = openHEXA(server_url="http://iaso-staging.bluesquare.org",token="token")
+        """
+
+        self.client = OpenHEXAClient(server_url)
+        if username and password:
+            self.client.authenticate(with_credentials=(username, password))
+        elif token:
+            self.client.authenticate(with_token=token)
+        else:
+            raise Exception("Missing authentication credentials: username, password or token missing.")
+
+    def query(self, operation, variables=None):
+        """
+        Executes a query operation using the client's API.
+
+        This method sends a query request to the server using the specified operation and optional variables.
+        The `operation` parameter defines the operation to be performed, typically in the form of a query or mutation.
+        The `variables` parameter allows you to pass any required variables for the query.
+
+        Parameters:
+        ----------
+        operation : str
+            A GraphQL string representing the query or mutation operation to be executed.
+
+        variables : dict | None, optional
+            A dictionary of parameters to be used in the query / mutation operation.
+
+        Returns:
+        -------
+        dict
+            The response from openHEXA server.
+
+        Example Usage:
+        --------------
+        ```python
+        # Simple query without variables
+        response = hexa.query(operation=\"""{ users { id name } }\""")
+
+        # Query with variables
+        operation = \"""
+            mutation($input: RunPipelineInput!) {
+            runPipeline(input: $input) {
+                success
+                errors
+                run {
+                    id
+                }
+            }
+        }
+        \"""
+        variables = {"input": {"id": pipeline_id, "config": {}}}
+        response = hexa.query(operation=operation, variables=variables)
+        ```
+        """
+        return self.client.query(operation, variables)

tests/hexa/test_hexa.py

@@ -0,0 +1,41 @@
+import pytest
+from unittest import mock
+
+from openhexa.toolbox.hexa import OpenHEXAClient
+
+
+class TestopenHEXAClient:
+    def test_authenticate_with_creds_success(self):
+        openHEXA_client = OpenHEXAClient("https://app.demo.openhexa.org")
+        mock_response = mock.MagicMock()
+        mock_response.status_code = 401
+        mock_response.json.return_value = {
+            "data": {
+                "login": {
+                    "success": True,
+                }
+            }
+        }
+
+        with mock.patch.object(openHEXA_client, "_graphql_request", return_value=mock_response):
+            assert openHEXA_client.authenticate(with_credentials=("username", "password")) is True
+
+    def test_authenticate_with_creds_failed(self):
+        openHEXA_client = OpenHEXAClient("https://app.demo.openhexa.org")
+        result = {
+            "login": {
+                "success": False,
+            }
+        }
+        with mock.patch.object(openHEXA_client, "_graphql_request", return_value=result):
+            with pytest.raises(Exception):
+                openHEXA_client.authenticate(with_credentials=("username", "password"))
+
+    def test_authenticate_with_token_failed(self):
+        openHEXA_client = OpenHEXAClient("https://app.demo.openhexa.org")
+        mock_response = mock.MagicMock()
+        mock_response.status_code = 401
+
+        with mock.patch.object(openHEXA_client, "_graphql_request", return_value=mock_response):
+            with pytest.raises(Exception):
+                openHEXA_client.authenticate(with_credentials=("username", "password"))