Flow Binary API

The Flow Binary API allows you to compile a flow, generate and run a flow binary, and list, pause, resume, cancel, or retry flow jobs.

In this document, URL_BASE refers to the root URL of your Instabase instance, such as https://www.instabase.com.

Compile a flow binary

Description

Compile a flow to generate a flow binary.

Request parameters

Parameters are required unless marked as optional.

Response schema

All keys are returned in the response by default, unless marked as optional.

Examples

Request

curl --location --request POST $URL_BASE'/api/v1/flow_binary/compile'
--header 'Authorization: Bearer '${ACCESS_TOKEN} \
--header 'Content-Type: application/json' \
--data-raw '{"binary_type": "Single Flow", \
"flow_project_root": "user1/HITL/fs/Instabase Drive/files/Checkpoint Step/", \
"settings": {"flow_file": "Workflows/checkpoint_step.ibflow"}}'

Response

{
    "status": "OK",
    "job_id": "6243b9a5-d1b7-4613-9acd-2e9571028853"
}

Run a Flow Binary

Description

Run a flow binary.

Request parameters

Parameters are required unless marked as optional.

Response schema

All keys are returned in the response by default, unless marked as optional.

Examples

Request

url = url_base + '/api/v1/flow/run_binary_async'

args = {
  'input_dir': "jaydoe/my_repo/fs/Instabase Drive/flow_proj/data/input",
  'binary_path': "jaydoe/my_repo/fs/Instabase Drive/flow_proj/build/bin/my_flow.ibflowbin",
  'settings': {
    'delete_out_dir': False,
    'output_has_run_id': True,
    'notification_emails': ["me@domain.com"],
  },
}
json_data = json.dumps(args)

headers = {
  'Authorization': 'Bearer {0}'.format(token)
}

r = requests.post(url, data=json_data, headers=headers)
resp_data = json.loads(r.content)

Response

The HTTP call returns immediately while the binary execution proceeds asynchronously in the background. If successful, the response contains a job_id field that you can use to check the status of your execution.

{
  "status": "OK",
  "data": {
    "job_id": "756be65b-0eaf-4192-bea5-176f0377b0f8",
    "output_folder": "jaydoe/my_repo/fs/Instabase Drive/flow_proj/data/out"
  }
}

Webhook Configuration

You can use the webhook_config setting to ensure your application is notified when a flow completes. Instabase POSTs JSON-encoded data of the format below to the webhook endpoint when the flow completes.

# body
{
  "status": <string>,
  "msg": <string>,
  "job_id": <string>,
  "binary_path": <string>,
  "input_dir": <string>,
  "output": <string>
}

The response body contains the following fields:

• status: "OK" | "ERROR"

• msg: (optional) Error message. Present only if status is ERROR.

• job_id: A unique identifier for the job.

• binary_path: The path reference to flow binary file.

• input_dir: Input directory.

• output: The full path to the root output folder.

To acknowledge receipt of the event, your endpoint must return a 2xx HTTP status code to Instabase. All response codes outside this range, including 3xx codes, indicate to Instabase that you did not receive the event.

If Instabase does not receive a 2xx HTTP status code, the notification attempt is repeated upto 7 times.

Add a log prefix

Add a prefix to all log statements generated by this flow run.

Pass the field log_prefix with Flow runtime config, where the value of log_prefix is the string that you want to prepend to all log statements. For example, using the runtime config below would prepend org=Piston to all log statements.

"runtime_config": {"log_prefix": "org=Piston "}

Change output schema at runtime

Change the output schema at runtime for Marketplace solutions and flow binary jobs.

Pass the field modify_output_schema with Flow runtime config, where the value of modify_output_schema is a stringify JSON object that can contain these fields:

• add_fields: a list of fields to add to the output. Each field is described as an object {"name": "field_name", "formula": "field_formula"}

• remove_fields: a list of field names to remove from the output

• rename_fields: an object where each key is the field to be renamed, and its value is the new name.

Here’s a JavaScript example of changing the output schema at runtime:

const output_config = {
  "add_fields": [{"name": "my_field_name", "formula": "echo(‘hello world’)"}],
  "remove_fields": ["field_1", "field_2"],
  "rename_fields": {"old_name_1": "new_name_1", "old_name_2": "new_name_2"}
}
const form = {
  "binary_path": binaryPath,
  "input_dir": inputDir,
    "settings": {
      "runtime_config": {"modify_output_schema": JSON.stringify(output_config)}
      ...
  }
}

$.ajax({
      type: 'POST',
      url: '/api/v1/flow/run_binary_async',
      headers: {
        'Content-Type': 'application/json;charset=UTF-8',
        'Authorization': 'Bearer '+token
      },
      data: JSON.stringify(form)
    })

Job Status

Use the Job Status API to get the execution status of a flow binary job.

Refer to the Job Status API.

Run a flow binary in sync mode

Description

Use the Sync API to synchronously run a Flow and return the output of the Flow in the specified format. The Sync API takes one or more files as input, runs the flow, blocks for the flow to finish and returns the extracted results. Flows run via Sync API run entirely in-memory and hence is faster than running the flow using the /run_binary_async API. However, the Sync API doesn’t support checkpoint resume, processing large batches (>5) of files, and has a max processing time of 60s.

Request parameters

Parameters are required unless marked as optional.

Response schema

All keys are returned in the response by default, unless marked as optional.

Examples

Request

import base64

encoded_string = base64.b64encode(open("test.pdf", "rb").read()).decode('utf-8')

url = url_base + '/api/v1/flow/run_binary_sync'

api_args = {
  'binary_path': "jaydoe/my_repo/fs/InstabaseDrive/build/bin/workflow/my_flow.ibflowbin",
  'files': [{
    'file_name': "test.pdf",
    'file_content': encoded_string,
  }]
}
json_data = json.dumps(api_args)

headers = {
  'Authorization': 'Bearer {0}'.format(token)
}

r = requests.post(url, data=json_data, headers=headers, verify=False)
resp_data = json.loads(r.content)

Response

{
  "records": [
    {
      "results": [
        {
          "key": "first_name",
          "value": "Coder"
        },
        {
          "key": "last_name",
          "value": "Maloder"
        }
      ],
      "file_name": "test.pdf",
      "file_index": 0,
      "record_index": 0
    }
  ],
  "binary_path": "jaydoe/my_repo/fs/InstabaseDrive/build/bin/workflow/my_flow.ibflowbin",
  "job_id": "5eb7df69-7506-41b3-bd3a-eb69d8b8a36d"
}