feat(openHEXA): add openHEXA client

docs/hexa.md

@@ -0,0 +1,45 @@
+
+# OpenHEXA Toolbox OpenHEXAClient
+
+The OpenHEXAClient module enables users to interact with the OpenHEXA backend using GraphQL syntax. 
+Its primary goal is to simplify communication with OpenHEXA and streamline integration with third-party applications.
+
+* [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")
+page=1,per_page=10
+
+workspaces = hexa.query("""
+    query {
+        workspaces (page: $page, perPage: $perPage) {
+            items {
+                slug
+                name
+            }
+        }
+    }
+""", {"page":page,"perPage":per_page})["workspaces"]["items"]
+print("Workspaces:")
+print(json.dumps(workspaces, indent=2))
+```
+
+
+

openhexa/toolbox/hexa/__init__.py

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

openhexa/toolbox/hexa/api.py

@@ -0,0 +1,67 @@
+from requests import Session
+import typing
+
+
+class NotFound(Exception):
+    """Errors related to an element not found."""
+
+    pass
+
+
+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"]