-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
6873358
commit be6422e
Showing
7 changed files
with
596 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,47 @@ | ||
|
||
# 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 but make two-factor authentication is disabled else it will not work | ||
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)) | ||
``` | ||
|
||
|
||
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,4 @@ | ||
from .api import OpenHEXAClient, NotFound | ||
from .hexa import OpenHEXA | ||
|
||
__all__ = ["OpenHEXA", "OpenHEXAClient", "NotFound"] |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,73 @@ | ||
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: | ||
message = ( | ||
"Authentication failed: login with token requires two-factor authentication to be disabled." | ||
if with_token | ||
else "Authentication failed." | ||
) | ||
|
||
raise Exception(message) | ||
|
||
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"] |
Oops, something went wrong.