hiro-graph-client
Advanced tools
This is a client library to access data of the HIRO Graph.
This library also contains classes for handling the WebSockets event-ws and action-ws API.
To process large batches of data, take a look at "hiro-batch-client". - ( PyPI: hiro-batch-client, GitHub: hiro-batch-client-python)
For more information about HIRO Automation, look at https://www.almato.com/de/loesungen/hiro-ai.
For more information about the APIs this library covers, see https://dev-portal.engine.datagroup.de/7.0/.
Currently, implemented are
HiroApp for appHiroAuth for authHiroGraph for graphHiroIam for iamHiroKi for kiHiroAuthz for authzHiroVariables for variablesand for the websockets
AbstractEventWebSocketHandler for event-wsAbstractActionWebSocketHandler for action-wsTo use this library, you will need an account at https://desktop.engine.datagroup.de/ and access to an OAuth Client-Id and Client-Secret to access the HIRO Graph. See also https://dev-portal.engine.datagroup.de/7.0/.
Most of the documentation is done in the sourcecode.
Example to use the straightforward graph api client:
from hiro_graph_client import PasswordAuthTokenApiHandler, HiroGraph
hiro_client: HiroGraph = HiroGraph(
api_handler=PasswordAuthTokenApiHandler(
root_url="https://core.engine.datagroup.de",
username='',
password='',
client_id='',
client_secret=''
)
)
# The commands of the Graph API are methods of the class HIROGraph.
# The next line executes a vertex query for instance.
query_result = hiro_client.query('ogit\\/_type:"ogit/MARS/Machine"')
print(query_result)
Authorization against the HIRO Graph is done via tokens. These tokens are handled by classes of
type AbstractTokenApiHandler in this library. Each of the Hiro-Client-Object (HiroGraph, , HiroApp, etc.) need to
have some kind of TokenApiHandler at construction.
This TokenApiHandler is also responsible to determine the most up-to-date endpoints for the API calls. You can supply a
custom list of endpoints by using the dict parameter custom_endpoints= on construction.
A custom list of headers can also be set via the dict parameter headers= in the constructor. These would update the
internal headers. Header names can be supplied in any upper/lower-case.
This library supplies the following TokenApiHandlers:
A simple TokenApiHandler that is generated with a preset-token at construction. Cannot update its token.
A TokenApiHandler that reads an environment variable (default is HIRO_TOKEN) from the runtime environment. Will only
update its token when the environment variable changes externally.
This TokenApiHandler logs into the HiroAuth backend and obtains a token from login credentials. This is also the only TokenApiHandler (so far) that automatically tries to renew a token from the backend when it has expired.
All code examples in this documentation can use these TokenApiHandlers interchangeably, depending on how such a token is provided.
The HiroGraph example from above with another customized TokenApiHandler:
from hiro_graph_client import EnvironmentTokenApiHandler, HiroGraph
hiro_client: HiroGraph = HiroGraph(
api_handler=EnvironmentTokenApiHandler(
root_url="https://core.engine.datagroup.de"
)
)
# The commands of the Graph API are methods of the class HIROGraph.
# The next line executes a vertex query for instance.
query_result = hiro_client.query('ogit\\/_type:"ogit/MARS/Machine"')
print(query_result)
Example with additional parameters:
from hiro_graph_client import EnvironmentTokenApiHandler, HiroGraph
hiro_client: HiroGraph = HiroGraph(
api_handler=EnvironmentTokenApiHandler(
root_url="https://core.engine.datagroup.de",
env_var='_TOKEN',
headers={
'X-Custom-Header': 'My custom value'
},
custom_endpoints={
"graph": "/api/graph/7.2",
"auth": "/api/auth/6.2"
},
client_name="HiroGraph (testing)" # Will be used in the header 'User-Agent'
)
)
# The commands of the Graph API are methods of the class HIROGraph.
# The next line executes a vertex query for instance.
query_result = hiro_client.query('ogit\\/_type:"ogit/MARS/Machine"')
print(query_result)
When you need to access multiple APIs of HIRO, share the TokenApiHandler between the API objects to avoid unnecessary
requests for token- and version-information against HIRO. The TokenApiHandler will share a requests.Session, token-
and version-request-handling, and the token string itself between them.
from hiro_graph_client import HiroGraph, HiroApp, PasswordAuthTokenApiHandler
hiro_api_handler = PasswordAuthTokenApiHandler(
root_url="https://core.engine.datagroup.de",
username='',
password='',
client_id='',
client_secret=''
)
hiro_client: HiroGraph = HiroGraph(
api_handler=hiro_api_handler
)
hiro_app_client: HiroApp = HiroApp(
api_handler=hiro_api_handler
)
You can also let TokenApiHandlers share a common connection session instead of letting each of them create their own. This might prove useful in a multithreading environment where tokens have to be set externally or change often (i.e. one token per user per thread). This also ensures, that version-requests happen only once when the connection is initialized.
Use the parameters pool_maxsize and pool_block to further tune the connection parameters for parallel access to
the backend. See requests Session Objects
and the Python documentation of requests.adapters.HTTPAdapter and GraphConnectionHandler.
from hiro_graph_client import HiroGraph, HiroApp, FixedTokenApiHandler, GraphConnectionHandler
connection_handler = GraphConnectionHandler(
root_url="https://core.engine.datagroup.de",
pool_maxsize=200, # Optional: Max pool of cached connections for this connection session
pool_block=True, # Optional: Do not allow more parallel connections than pool_maxsize
client_name="Your Graph Client 0.0.1" # Optional: Will be used in the header 'User-Agent'
)
# Work with token of user 1
user1_client: HiroGraph = HiroGraph(
api_handler=FixedTokenApiHandler(
connection_handler=connection_handler,
token='token user 1'
)
)
# Work with token of user 2 (Shared Token Handler)
user2_api_handler = FixedTokenApiHandler(
connection_handler=connection_handler,
token='token user 2'
)
user2_client: HiroGraph = HiroGraph(
api_handler=user2_api_handler
)
hiro_app_client: HiroApp = HiroApp(
api_handler=user2_api_handler
)
Everything written in Token Handler Sharing still applies.
SSL parameters are configured using the class SSLConfig. This class translates the parameters given to the required
fields for the requests library of Python (parameters cert and verify there). This configuration is given to the
TokenApiHandlers and will be used by the clients attached to it as well.
If this is not set, the default settings of the library requests will be used, which is to verify any server
certificates by using system defaults.
from hiro_graph_client import EnvironmentTokenApiHandler, HiroGraph, SSLConfig
hiro_client: HiroGraph = HiroGraph(
api_handler=EnvironmentTokenApiHandler(
root_url="https://core.engine.datagroup.de",
# Disable any verification.
ssl_config=SSLConfig(verify=False)
)
)
query_result = hiro_client.query('ogit\\/_type:"ogit/MARS/Machine"')
print(query_result)
from hiro_graph_client import EnvironmentTokenApiHandler, HiroGraph, SSLConfig
hiro_client: HiroGraph = HiroGraph(
api_handler=EnvironmentTokenApiHandler(
root_url="https://core.engine.datagroup.de",
# Set custom certification files. If any of them are omitted, system defaults will be used.
ssl_config=SSLConfig(
cert_file="<path to client certificate file>",
key_file="<path to key file for the client certificate>",
ca_bundle_file="<path to the ca_bundle to verify the server certificate>"
)
)
)
query_result = hiro_client.query('ogit\\/_type:"ogit/MARS/Machine"')
print(query_result)
The Graph Client is mostly straightforward to use, since all public methods of this class represent an API call in the Graph API. Documentation is available in source code as well. Some calls are a bit more complicated though and explained in more detail below:
To upload data to such a vertex, use HiroGraph.post_attachment(data=...). The parameter data= will be given directly
to the call of the Python library requests as requests.post(data=...). To stream data, set data to an object of
type IO. See the documentation of the Python library requests for more details.
Downloading an attachment is done in chunks, so huge blobs of data in memory can be avoided when streaming this data. Each chunk is 64k by default.
To stream such an attachment to a file, see the example below:
ogit_id = '<ogit/_id of a vertex>'
data_iter = hiro_client.get_attachment(ogit_id)
with io.start("attachment.bin", "wb") as file:
for chunk in data_iter:
file.write(chunk)
To read the complete data in memory, see this example:
ogit_id = '<ogit/_id of a vertex>'
data_iter = hiro_client.get_attachment(ogit_id)
attachment = b''.join(data_iter)
This library contains classes that make using HIRO WebSocket protocols easier. They handle authentication, exceptions and much more.
The classes do not handle buffering of messages, so it is the duty of the programmer to ensure, that incoming messages are either handled quickly or being buffered to avoid clogging the websocket. The classes are thread-safe, so it is possible to handle each incoming message asynchronously in its own thread and have those threads send results back if needed (See multithreaded example in Action WebSocket).
To shut these WebSockets (ws) down cleanly, please consider these scenarios:
The library reacts on KeyboardInterrupt (SIGINT) and closes the WebSocket cleanly with closing message to the sever when such an interrupt is received. If another signal (like SIGTERM) is received, the program will stop immediately without any closing message back to the server.
When installing signal handlers, you need to use ws.signal_stop() to shut the WebSocket down. Do NOT use ws.stop() or the closing process will deadlock. This is because the signal interrupt is executed in the same thread as ws.run_forever().
Example:
import signal
[...]
with ActionWebSocket(api_handler=FixedTokenApiHandler('HIRO_TOKEN')) as ws:
def signal_handler(signum, handler):
ws.signal_stop()
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGHUP, signal_handler) # This signal might not be available on MS Windows
ws.run_forever()
When closing the WebSocket from another thread than the one running ws.run_forever(), you should use ws.stop(). This ensures, that the shutdown is synchronized and ws.stop() will return after the WebSocket has been closed.
This websocket receives notifications about changes to vertices that match a certain filter.
See also API description of event-ws
Example:
import threading
from hiro_graph_client.clientlib import FixedTokenApiHandler
from hiro_graph_client.eventswebsocket import AbstractEventsWebSocketHandler, EventMessage, EventsFilter
class EventsWebSocket(AbstractEventsWebSocketHandler):
def on_create(self, message: EventMessage):
""" Vertex has been created """
print("Create:\n" + str(message))
def on_update(self, message: EventMessage):
""" Vertex has been updated """
print("Update:\n" + str(message))
def on_delete(self, message: EventMessage):
""" Vertex has been removed """
print("Delete:\n" + str(message))
events_filter = EventsFilter(filter_id='testfilter', filter_content="(element.ogit/_type=ogit/MARS/Machine)")
with EventsWebSocket(api_handler=FixedTokenApiHandler('HIRO_TOKEN'),
events_filters=[events_filter],
query_params={"allscopes": "false", "delta": "false"}) as ws:
ws.run_forever() # Use KeyboardInterrupt (Ctrl-C) to exit.
If you do not set the parameter scope=, the default scope of your account will be used. If you need to set the scope
by hand, use the following:
[...]
api_handler = FixedTokenApiHandler('HIRO_TOKEN')
default_scope = api_handler.decode_token()['data']['default-scope']
with EventsWebSocket(api_handler=api_handler,
events_filters=[events_filter],
scopes=[default_scope],
query_params={"allscopes": "false", "delta": "false"}) as ws:
ws.run_forever() # Use KeyboardInterrupt (Ctrl-C) to exit.
This websocket receives notifications about actions that have been triggered within a KI. Use this to write your own custom action handler.
See also API description of action-ws
Simple example:
import threading
from hiro_graph_client.actionwebsocket import AbstractActionWebSocketHandler
from hiro_graph_client.clientlib import FixedTokenApiHandler
class ActionWebSocket(AbstractActionWebSocketHandler):
def on_submit_action(self, action_id: str, capability: str, parameters: dict):
""" Message *submitAction* has been received """
# Handle the message
print(f"ID: {action_id}, Capability: {capability}, Parameters: {str(parameters)}")
# Send back message *sendActionResult*
self.send_action_result(action_id, "Everything went fine.")
def on_config_changed(self):
""" The configuration of the ActionHandler has changed """
pass
with ActionWebSocket(api_handler=FixedTokenApiHandler('HIRO_TOKEN')) as ws:
ws.run_forever() # Use KeyboardInterrupt (Ctrl-C) to exit.
Multithreading example using a thread executor:
import threading
import concurrent.futures
from hiro_graph_client.actionwebsocket import AbstractActionWebSocketHandler
from hiro_graph_client.clientlib import FixedTokenApiHandler, AbstractTokenApiHandler
class ActionWebSocket(AbstractActionWebSocketHandler):
def __init__(self, api_handler: AbstractTokenApiHandler):
""" Initialize properties """
super().__init__(api_handler)
self._executor = None
def start(self) -> None:
""" Initialize the executor """
super().start()
self._executor = concurrent.futures.ThreadPoolExecutor()
def stop(self, timeout: int = None) -> None:
""" Shut the executor down """
if self._executor:
self._executor.shutdown()
self._executor = None
super().stop(timeout)
def handle_submit_action(self, action_id: str, capability: str, parameters: dict):
""" Runs asynchronously in its own thread. """
print(f"ID: {action_id}, Capability: {capability}, Parameters: {str(parameters)}")
self.send_action_result(action_id, "Everything went fine.")
def on_submit_action(self, action_id: str, capability: str, parameters: dict):
""" Message *submitAction* has been received. Message is handled in thread executor. """
if not self._executor:
raise RuntimeError('ActionWebSocket has not been started.')
self._executor.submit(ActionWebSocket.handle_submit_action, self, action_id, capability, parameters)
def on_config_changed(self):
""" The configuration of the ActionHandler has changed """
pass
with ActionWebSocket(api_handler=FixedTokenApiHandler('HIRO_TOKEN')) as ws:
ws.run_forever() # Use KeyboardInterrupt (Ctrl-C) to exit.
HiroGraph.escaped_query method.decode_token_ext().Changes for /api/auth/6.6:
access_token and _TOKEN in token results.fresh() from class TokenInfo because each token refresh issues a new refresh_token by default
now.GraphConnectonHandler which can be shared
between TokenApiHandlers and API-Classes.pool_connections=1, pool_maxsize=10 by default (the latter can be changed).pool_block to block any connections that would exceed the pool_maxsize.accept_all_certs().requests.Session and the pool_maxsize to documentation.batchclient.py as its own project at https://github.com/arago/hiro-batch-client-python.handle_vertices_combined recognizes _issue_data now for issues directly linked to
vertices.Refactoring of websocketlib.py:
Allow - or _ as build number separator on test tags, i.e. t4.8.0_0 and t4.8.0-0 are handled identical now.
self.submitStore and self.resultStore properly on shutdown in AbstractActionWebSocketHandler.
They might not have been initialized at all when startup fails.HiroGraphBatch. Removed obsolete methods.Updated some code documentation in HiroGraph that became obsolete.
Updated graph queries in HiroGraph:
count to vertex query.Renamed default _client_name to hiro-graph-client (name of the lib on PyPI).
Content- / Media-Type handling
WrongContentTypeError when an unexpected (or missing) Content-Type is returned.Error handling
ki.log_communication_on_error to enable logging of communication when an error is detected in the response.Add parameter max_tries to TokenHandlers.
BUGFIX
decode_token() to decode the information embedded in a HIRO token.python-hiro-clients to hiro-client-python. No other technical changes.query_params={'allscopes': 'true'} if you want to enable it for EventWebSockets. If this
is left out of the query_params, it will be added as 'allscopes': 'false'.query_params= for WebSockets is of type Dict[str, str].query_params={'allscopes': 'false'} as default for the EventWebSocket.query_params to add arbitrary query parameters to the initial websocket request.Changes to AbstractAuthenticatedWebSocketHandler:
Introducing run_forever() which will return after the reader thread has been joined. This can happen when another
thread calls stop() (normal exit) or an internal error occurs, either directly when the connection is attempted, a
token got invalid and could not be refreshed, or any other exception that has been thrown in the internal reader
thread.
This ensures, that the main thread does not continue when the websocket reader thread is not there.
Enable parallel executions of send() across multiple threads.
Make sure, that only one thread triggers a restart by a call to restart().
Check for active websocket reader thread via is_active().
Update examples for websockets in README.md.
Generic
client_name.client_name to give connections a name and also set header User-Agent more easily.remote_exit_codes to AbstractAuthenticatedWebSocketHandler._backoff() by not using sleep() but threading.Condition.wait().handle_vertices_combinedAbstractTokenApiHandler
AbstractIOCarrier works with with statements now.
Added BasicFileIOCarrier.
Removed ApiConfig.
Renamed several internal classes.
Better error messages.
HTTP secure protocol logging.
Fixed timestamp creation for tokens.
Joe's suggestions - thanks Joe!
Added VERSION to package_data in setup.py
Added documentation for PyPI
Initial release after split from https://github.com/arago/hiro-clients
FAQs
Hiro Client for Graph REST API of HIRO 7
We found that hiro-graph-client demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 3 open source maintainers collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
GitHub postponed a new billing model for self-hosted Actions after developer pushback, but moved forward with hosted runner price cuts on January 1.
Research
Destructive malware is rising across open source registries, using delays and kill switches to wipe code, break builds, and disrupt CI/CD.
Security News
Socket CTO Ahmad Nassri shares practical AI coding techniques, tools, and team workflows, plus what still feels noisy and why shipping remains human-led.