Bases: object
A low level client to invoke api methods from Lucidtech AI Services.
Creates an asset, calls the POST /assets endpoint.
>>> from las.client import Client>>> client = Client()>>> client.create_asset(b'<bytes data>')
Parameters
content (Content) – Content to POST
name (Optional[str**]) – Name of the asset
description (Optional[str**]) – Description of the asset
Returns
Asset response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Creates a batch, calls the POST /batches endpoint.
>>> from las.client import Client>>> client = Client()>>> client.create_batch(description='<description>')
Parameters
name (Optional[str**]) – Name of the batch
description (Optional[str**]) – Description of the batch
Returns
Batch response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Creates a document, calls the POST /documents endpoint.
>>> from las.client import Client>>> client = Client()>>> client.create_document(b'<bytes data>', 'image/jpeg', consent_id='<consent id>')
Parameters
content (Content) – Content to POST
content_type (str) – MIME type for the document
consent_id (Optional[str**]) – Id of the consent that marks the owner of the document
batch_id (Optional[str**]) – Id of the associated batch
ground_truth (Optional[Sequence[Dict[str, str]]**]) – List of items {label: value} representing the ground truth values for the document
Returns
Document response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Create a prediction on a document using specified model, calls the POST /predictions endpoint.
>>> from las.client import Client>>> client = Client()>>> client.create_prediction(document_id='<document id>', model_id='<model id>')
Parameters
document_id (str) – Id of the document to run inference and create a prediction on
model_id (str) – Id of the model to use for inference
max_pages (Optional[int**]) – Maximum number of pages to run predictions on
auto_rotate (Optional[bool**]) – Whether or not to let the API try different rotations on the document when running predictions
image_quality (Optional[int**]) – image quality for prediction “LOW|HIGH”. high quality could give better result but will also take longer time.
Returns
Prediction response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Creates an secret, calls the POST /secrets endpoint.
>>> from las.client import Client>>> client = Client()>>> data = {'username': '<username>', 'password': '<password>'}>>> client.create_secret(data, description='<description>')
Parameters
data (str) – Dict containing the data you want to keep secret
name (Optional[str**]) – Name of the secret
description (Optional[str**]) – Description of the secret
Returns
Secret response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Creates a transition, calls the POST /transitions endpoint.
>>> import json>>> from pathlib import Path>>> from las.client import Client>>> client = Client()>>> in_schema = {'$schema': 'https://json-schema.org/draft-04/schema#', 'title': 'in', 'properties': {...} }>>> out_schema = {'$schema': 'https://json-schema.org/draft-04/schema#', 'title': 'out', 'properties': {...} }>>> # A typical docker transition>>> docker_params = {>>> 'imageUrl': '<image_url>',>>> 'credentials': {'username': '<username>', 'password': '<password>'}>>> }>>> client.create_transition('docker', in_schema=in_schema, out_schema=out_schema, params=docker_params)>>> # A manual transition with UI>>> assets = {'jsRemoteComponent': 'las:asset:<hex-uuid>', '<other asset name>': 'las:asset:<hex-uuid>'}>>> manual_params = {'assets': assets}>>> client.create_transition('manual', in_schema=in_schema, out_schema=out_schema, params=manual_params)
Parameters
transition_type (str) – Type of transition “docker”|”manual”
in_schema (Optional[dict**]) – Json-schema that defines the input to the transition
out_schema (Optional[dict**]) – Json-schema that defines the output of the transition
name (Optional[str**]) – Name of the transition
parameters (Optional[dict**]) – Parameters to the corresponding transition type
description (Optional[str**]) – Description of the transition
Returns
Transition response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Creates a new user, calls the POST /users endpoint.
>>> from las.client import Client>>> client = Client()>>> client.create_user('<email>', name='John Doe')
Parameters
email (str) – Email to the new user
name (Optional[str**]) – Name of the user
avatar (Optional[str**]) – base64 encoded JPEG avatar of the user
Returns
User response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Creates a new workflow, calls the POST /workflows endpoint. Check out Lucidtech’s tutorials for more info on how to create a workflow.
>>> from las.client import Client>>> from pathlib import Path>>> client = Client()>>> specification = {'language': 'ASL', 'version': '1.0.0', 'definition': {...}}>>> error_config = {'email': '<error-recipient>'}>>> client.create_workflow(specification, error_config=error_config)
Parameters
specification (dict) – Specification of the workflow,
currently supporting ASL: https://states-language.net/spec.html
name (Optional[str**]) – Name of the workflow
description (Optional[str**]) – Description of the workflow
error_config (Optional[dict**]) – Configuration of error handler
completed_config (Optional[dict**]) – Configuration of a job to run whenever a workflow execution ends
Returns
Workflow response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Delete the asset with the provided asset_id, calls the DELETE /assets/{assetId} endpoint.
>>> from las.client import Client>>> client = Client()>>> client.delete_asset('<asset_id>')
Parameters
asset_id (str) – Id of the asset
Returns
Asset response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Delete documents with the provided consent_id, calls the DELETE /documents endpoint.
>>> from las.client import Client>>> client = Client()>>> client.delete_documents(consent_id='<consent id>')
Parameters
consent_id (Optional[Queryparam**]) – Ids of the consents that marks the owner of the document
max_results (Optional[int**]) – Maximum number of documents that will be deleted
next_token (Optional[str**]) – A unique token for each page, use the returned token to retrieve the next page.
Returns
Documents response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Delete the secret with the provided secret_id, calls the DELETE /secrets/{secretId} endpoint.
>>> from las.client import Client>>> client = Client()>>> client.delete_secret('<secret_id>')
Parameters
secret_id (str) – Id of the secret
Returns
Secret response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Delete the transition with the provided transition_id, calls the DELETE /transitions/{transitionId} endpoint.
Will fail if transition is in use by one or more workflows.
>>> from las.client import Client>>> client = Client()>>> client.delete_transition('<transition_id>')
Parameters
transition_id (str) – Id of the transition
Returns
Transition response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Delete the user with the provided user_id, calls the DELETE /users/{userId} endpoint.
>>> from las.client import Client>>> client = Client()>>> client.delete_user('<user_id>')
Parameters
user_id (str) – Id of the user
Returns
User response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Delete the workflow with the provided workflow_id, calls the DELETE /workflows/{workflowId} endpoint.
>>> from las.client import Client>>> client = Client()>>> client.delete_workflow('<workflow_id>')
Parameters
workflow_id (str) – Id of the workflow
Returns
Workflow response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Deletes the execution with the provided execution_id from workflow_id, calls the DELETE /workflows/{workflowId}/executions/{executionId} endpoint.
>>> from las.client import Client>>> client = Client()>>> client.delete_workflow_execution('<workflow_id>', '<execution_id>')
Parameters
workflow_id (str) – Id of the workflow
execution_id (str) – Id of the execution
Returns
WorkflowExecution response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Start executing a manual transition, calls the POST /transitions/{transitionId}/executions endpoint.
>>> from las.client import Client>>> client = Client()>>> client.execute_transition('<transition_id>')
Parameters
transition_id (str) – Id of the transition
Returns
Transition execution response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Start a workflow execution, calls the POST /workflows/{workflowId}/executions endpoint.
>>> from las.client import Client>>> from pathlib import Path>>> client = Client()>>> content = {...}>>> client.execute_workflow('<workflow_id>', content)
Parameters
workflow_id (str) – Id of the workflow
content (dict) – Input to the first step of the workflow
Returns
Workflow execution response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Get asset, calls the GET /assets/{assetId} endpoint.
>>> from las.client import Client>>> client = Client()>>> client.get_asset(asset_id='<asset id>')
Parameters
asset_id (str) – Id of the asset
Returns
Asset response from REST API with content
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Get document, calls the GET /documents/{documentId} endpoint.
>>> from las.client import Client>>> client = Client()>>> client.get_document('<document id>')
Parameters
document_id (str) – Id of the document
Returns
Document response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
get log, calls the GET /logs/{logId} endpoint.
>>> from las.client import Client>>> client = Client()>>> client.get_log('<log_id>')
Parameters
log_id (str) – Id of the log
Returns
Log response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Get the transition with the provided transition_id, calls the GET /transitions/{transitionId} endpoint.
>>> from las.client import Client>>> client = Client()>>> client.get_transition('<transition_id>')
Parameters
transition_id (str) – Id of the transition
Returns
Transition response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Get an execution of a transition, calls the GET /transitions/{transitionId}/executions/{executionId} endpoint
>>> from las.client import Client>>> client = Client()>>> client.get_transition_execution('<transition_id>', '<execution_id>')
Parameters
transition_id (str) – Id of the transition
execution_id (str) – Id of the executions
Returns
Transition execution responses from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Get information about a specific user, calls the GET /users/{user_id} endpoint.
>>> from las.client import Client>>> client = Client()>>> client.get_user('<user_id>')
Parameters
user_id (str) – Id of the user
Returns
User response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Get the workflow with the provided workflow_id, calls the GET /workflows/{workflowId} endpoint.
>>> from las.client import Client>>> client = Client()>>> client.get_workflow('<workflow_id>')
Parameters
workflow_id (str) – Id of the workflow
Returns
Workflow response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Get a workflow execution, calls the GET /workflows/{workflow_id}/executions/{execution_id} endpoint.
>>> from las.client import Client>>> client = Client()>>> client.get_workflow_execution('<workflow_id>', '<execution_id>')
Parameters
workflow_id (str) – Id of the workflow that performs the execution
execution_id (str) – Id of the execution to get
Returns
Workflow execution response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
List assets available, calls the GET /assets endpoint.
>>> from las.client import Client>>> client = Client()>>> client.list_assets()
Parameters
max_results (Optional[int**]) – Maximum number of results to be returned
next_token (Optional[str**]) – A unique token for each page, use the returned token to retrieve the next page.
Returns
Assets response from REST API without the content of each asset
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
List documents available for inference, calls the GET /documents endpoint.
>>> from las.client import Client>>> client = Client()>>> client.list_documents(batch_id='<batch_id>', consent_id='<consent_id>')
Parameters
batch_id (Optional[Queryparam**]) – Ids of batches that contains the documents of interest
consent_id (Optional[Queryparam**]) – Ids of the consents that marks the owner of the document
max_results (Optional[int**]) – Maximum number of results to be returned
next_token (Optional[str**]) – A unique token for each page, use the returned token to retrieve the next page.
Returns
Documents response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
List models available, calls the GET /models endpoint.
>>> from las.client import Client>>> client = Client()>>> client.list_models()
Parameters
max_results (Optional[int**]) – Maximum number of results to be returned
next_token (Optional[str**]) – A unique token for each page, use the returned token to retrieve the next page.
Returns
Models response from REST API without the content of each model
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
List predictions available, calls the GET /predictions endpoint.
>>> from las.client import Client>>> client = Client()>>> client.list_predictions()
Parameters
max_results (Optional[int**]) – Maximum number of results to be returned
next_token (Optional[str**]) – A unique token for each page, use the returned token to retrieve the next page.
Returns
Predictions response from REST API without the content of each prediction
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
List secrets available, calls the GET /secrets endpoint.
>>> from las.client import Client>>> client = Client()>>> client.list_secrets()
Parameters
max_results (Optional[int**]) – Maximum number of results to be returned
next_token (Optional[str**]) – A unique token for each page, use the returned token to retrieve the next page.
Returns
Secrets response from REST API without the username of each secret
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
List executions in a transition, calls the GET /transitions/{transitionId}/executions endpoint.
>>> from las.client import Client>>> client = Client()>>> client.list_transition_executions('<transition_id>', '<status>')
Parameters
transition_id (str) – Id of the transition
status (Optional[Queryparam**]) – Statuses of the executions
order (Optional[str**]) – Order of the executions, either ‘ascending’ or ‘descending’
sort_by (Optional[str**]) – the sorting variable of the executions, either ‘endTime’, or ‘startTime’
execution_id (Optional[Queryparam**]) – Ids of the executions
max_results (Optional[int**]) – Maximum number of results to be returned
next_token (Optional[str**]) – A unique token for each page, use the returned token to retrieve the next page.
Returns
Transition executions responses from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
List transitions, calls the GET /transitions endpoint.
>>> from las.client import Client>>> client = Client()>>> client.list_transitions('<transition_type>')
Parameters
transition_type (Optional[Queryparam**]) – Types of transitions
max_results (Optional[int**]) – Maximum number of results to be returned
next_token (Optional[str**]) – A unique token for each page, use the returned token to retrieve the next page.
Returns
Transitions response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
List users, calls the GET /users endpoint.
>>> from las.client import Client>>> client = Client()>>> client.list_users()
Parameters
max_results (Optional[int**]) – Maximum number of results to be returned
next_token (Optional[str**]) – A unique token for each page, use the returned token to retrieve the next page.
Returns
Users response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
List executions in a workflow, calls the GET /workflows/{workflowId}/executions endpoint.
>>> from las.client import Client>>> client = Client()>>> client.list_workflow_executions('<workflow_id>', '<status>')
Parameters
workflow_id (str) – Id of the workflow
order (Optional[str**]) – Order of the executions, either ‘ascending’ or ‘descending’
sort_by (Optional[str**]) – the sorting variable of the executions, either ‘endTime’, or ‘startTime’
status (Optional[Queryparam**]) – Statuses of the executions
max_results (Optional[int**]) – Maximum number of results to be returned
next_token (Optional[str**]) – A unique token for each page, use the returned token to retrieve the next page.
Returns
Workflow executions responses from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
List workflows, calls the GET /workflows endpoint.
>>> from las.client import Client>>> client = Client()>>> client.list_workflows()
Parameters
max_results (Optional[int**]) – Maximum number of results to be returned
next_token (Optional[str**]) – A unique token for each page, use the returned token to retrieve the next page.
Returns
Workflows response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Send heartbeat for a manual execution to signal that we are still working on it. Must be done at minimum once every 60 seconds or the transition execution will time out, calls the POST /transitions/{transitionId}/executions/{executionId}/heartbeats endpoint.
>>> from las.client import Client>>> client = Client()>>> client.send_heartbeat('<transition_id>', '<execution_id>')
Parameters
transition_id (str) – Id of the transition
execution_id (str) – Id of the transition execution
Returns
Empty response
Return type
None
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Updates an asset, calls the PATCH /assets/{assetId} endpoint.
>>> from las.client import Client>>> client = Client()>>> client.update_asset('<asset id>', content=b'<bytes data>')
Parameters
asset_id (str) – Id of the asset
content (Optional[Content**]) – Content to PATCH
name (Optional[str**]) – Name of the asset
description (Optional[str**]) – Description of the asset
Returns
Asset response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Update ground truth for a document, calls the PATCH /documents/{documentId} endpoint. Updating ground truth means adding the ground truth data for the particular document. This enables the API to learn from past mistakes.
>>> from las.client import Client>>> client = Client()>>> ground_truth = [{'label': 'total_amount', 'value': '156.00'}, {'label': 'date', 'value': '2018-10-23'}]>>> client.update_document(document_id='<document id>', ground_truth=ground_truth)
Parameters
document_id (str) – Id of the document
ground_truth (Sequence[Dict[str, Union[Optional[str], bool]]]) – List of items {label: value} representing the ground truth values for the document
Returns
Document response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Updates an secret, calls the PATCH /secrets/secretId endpoint.
>>> from las.client import Client>>> client = Client()>>> data = {'username': '<username>', 'password': '<password>'}>>> client.update_secret('<secret id>', data, description='<description>')
Parameters
secret_id (str) – Id of the secret
data (Optional[dict**]) – Dict containing the data you want to keep secret
name (Optional[str**]) – Name of the secret
description (Optional[str**]) – Description of the secret
Returns
Secret response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Updates a transition, calls the PATCH /transitions/{transitionId} endpoint.
>>> import json>>> from pathlib import Path>>> from las.client import Client>>> client = Client()>>> client.update_transition('<transition-id>', name='<name>', description='<description>')
Parameters
transition_id (str) – Id of the transition
name (Optional[str**]) – Name of the transition
description (Optional[str**]) – Description of the transition
in_schema (Optional[dict**]) – Json-schema that defines the input to the transition
out_schema (Optional[dict**]) – Json-schema that defines the output of the transition
assets (Optional[dict**]) – A dictionary where the values are assetIds that can be used in a manual transition
environment (Optional[dict**]) – Environment variables to use for a docker transition
environment_secrets –
A list of secretIds that contains environment variables to use for a docker transition :type environment_secrets: Optional[list] :return: Transition response from REST API :rtype: dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Ends the processing of the transition execution, calls the PATCH /transitions/{transition_id}/executions/{execution_id} endpoint.
>>> from las.client import Client>>> client = Client()>>> output = {...}>>> client.update_transition_execution('<transition_id>', '<execution_id>', 'succeeded', output)>>> error = {"message": 'The execution could not be processed due to ...'}>>> client.update_transition_execution('<transition_id>', '<execution_id>', 'failed', error)
Parameters
transition_id (str) – Id of the transition that performs the execution
execution_id (str) – Id of the execution to update
status (str) – Status of the execution ‘succeeded|failed’
output (Optional[dict**]) – Output from the execution, required when status is ‘succeded’
error (Optional[dict**]) – Error from the execution, required when status is ‘failed’, needs to contain ‘message’
start_time (Optional[str**]) – start time that will replace the original start time of the execution
Returns
Transition execution response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Updates a user, calls the PATCH /users/{userId} endpoint.
>>> from las.client import Client>>> client = Client()>>> client.update_user('<user id>', name='John Doe')
Parameters
user_id (str) – Id of the user
name (Optional[str**]) – Name of the user
avatar (Optional[str**]) – base64 encoded JPEG avatar of the user
Returns
User response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Updates a workflow, calls the PATCH /workflows/{workflowId} endpoint.
>>> import json>>> from pathlib import Path>>> from las.client import Client>>> client = Client()>>> client.update_workflow('<workflow-id>', name='<name>', description='<description>')
Parameters
workflow_id – Id of the workflow
name (Optional[str**]) – Name of the workflow
description (Optional[str**]) – Description of the workflow
error_config (Optional[dict**]) – Configuration of error handler
completed_config (Optional[dict**]) – Configuration of a job to run whenever a workflow execution ends
Returns
Workflow response from REST API
Return type
dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Retry or end the processing of a workflow execution, calls the PATCH /workflows/{workflow_id}/executions/{execution_id} endpoint.
>>> from las.client import Client>>> client = Client()>>> client.update_workflow_execution('<workflow_id>', '<execution_id>', '<next_transition_id>')
Parameters
workflow_id (str) – Id of the workflow that performs the execution
execution_id (str) – Id of the execution to update
next_transition_id – the next transition to transition into, to end the workflow-execution,
use: las:transition:commons-failed :type next_transition_id: str :return: Workflow execution response from REST API :rtype: dict
Raises
InvalidCredentialsException
, TooManyRequestsException
, LimitExceededException
, requests.exception.RequestException
Bases: object
Used to fetch and store credentials and to generate/cache an access token.
Parameters
client_id – The client id
client_secret – The client secret
api_key – The api key
auth_endpoint – The auth endpoint
api_endpoint – The api endpoint