Flow Creation - Precisely Data Integrity Suite

What is a Flow

A Flow is a JSON-defined execution graph that orchestrates a series of operations (called operators) to process data. Each operator performs a specific task and connects to the next operator, forming a chain of operations.

• Flow: A collection of interconnected operators forming an execution graph.
• Operator: A node in the graph that performs a specific operation (transform data, make HTTP calls, branch logic, etc.).
• Global Context: A JSON object that holds all data during flow execution, accessible to all operators.
• JSONata: An expression language used for querying and transforming JSON data within operators.

Flow Structure

A flow definition in JSON consists of the following top-level properties:

{
  "flowId": "unique-flow-identifier",
  "startOperatorId": "first-operator-id",
  "inputDataKey": "request",
  "operators": [
    // Array of operator definitions
      ]
}
      

Flow Execution with Metadata

Metadata is an optional JSON object that can be passed when executing a flow. It provides contextual information that can be accessed by any operator in the flow through JSONata expressions. Metadata is separate from the input data and is typically used for configuration values, request tracking, user context, feature flags, and environment-specific parameters.

Flow Execution Request Structure:

{
  "metadata": {
    "key1": "value1",
    "key2": "value2"
  },
  "input": [
    {
      "data_field1": "data_value1",
      "data_field2": "data_value2"
    },
    {
      "data_field1": "data_value3",
      "data_field2": "data_value4"
    }
  ]
}
      

Accessing Metadata in Flows: Metadata can be accessed in JSONata expressions using the metadata prefix, e.g. metadata.geocodeUrl, metadata.requestId.

Metadata vs Input Data:

Example: Complete Flow with Metadata

Execution Request:

{
  "metadata": {
    "datasetType": "Premium",
    "requestId": "req-98765",
    "enableCaching": true
      },
  "input": [
    {
      "addressLines": ["123 Main St"],
      "city": "New York",
      "country": "USA"
    }
  ]
}
      

Flow Definition:

{
  "flowId": "geocode-with-metadata",
  "startOperatorId": "validate_country",
  "inputDataKey": "request",
  "operators": [
    {
      "id": "validate_country",
      "type": "Predicate",
      "description": "Check if country is supported",
      "condition": "$exists(request.country) and request.country != ''",
      "successNode": "prepare_request",
      "failureNode": "country_missing_error"
    },
    {
      "id": "prepare_request",
      "type": "Transform",
      "description": "Build geocode request with metadata parameters",
      "operation": "{'address': request, 'dataset': metadata.datasetType, 'requestId': metadata.requestId}",
      "key": "geocode_request",
      "nextNode": "call_geocode"
    },
    {
      "id": "call_geocode",
      "type": "Post",
      "description": "Call geocoding API using metadata URL",
      "url": "https://api.precisely.com/v1/geocode",
      "headers": "{'Content-Type': 'application/json', 'X-Request-ID': metadata.requestId}",
      "body": "{'addresses': [request], 'preferences': {'customPreferences': {'DATA_OPTION': metadata.datasetType}}}",
      "requiredElements": "responses[0]",
      "key": "geocode_result",
      "nextNode": "check_cache_flag"
    },
    {
      "id": "check_cache_flag",
      "type": "Predicate",
      "description": "Check if caching should be enabled",
      "condition": "metadata.enableCaching = true",
      "successNode": "return_with_cache_info",
      "failureNode": "return_result"
    },
    {
      "id": "return_with_cache_info",
      "type": "Return",
      "resultExpression": "{'result': geocode_result, 'cached': true, 'requestId': metadata.requestId}"
    },
    {
      "id": "return_result",
      "type": "Return",
      "resultExpression": "{'result': geocode_result, 'requestId': metadata.requestId}"
    },
    {
      "id": "country_missing_error",
      "type": "Fail",
      "errorCode": 400,
      "status": "VALIDATION_ERROR",
      "errorMessage": "Country field is required"
    }
  ]
}
      

Handling Optional Metadata: Always check for existence using $exists(metadata.field) and provide defaults as needed.

$exists(metadata.apiUrl) ? metadata.apiUrl : 'https://default-api.example.com'
$exists(metadata.usePremium) and metadata.usePremium = true
metadata.timeout ? metadata.timeout : 5000
      

Available Operators

The Multipass SDK provides several built-in operators for common operations. Each operator has a specific purpose and set of properties.

• Simple Operators: Perform operations and proceed to a single next node (Transform, Get, Post)
• Conditional Operators: Branch execution based on conditions (Predicate)
• Termination Operators: End flow execution and return results (Return, Fail)

Transform Operator

Purpose: Transform data using JSONata expressions and store the result in the global context.

Use Cases: Extract fields, perform calculations, restructure JSON, combine fields.

JSON Example:

{
  "id": "transform_city",
  "type": "Transform",
  "description": "Convert city name to uppercase and trim spaces",
  "operation": "$uppercase($trim(request.city))",
  "key": "normalized_city",
  "nextNode": "next_operator"
}
      

Common JSONata Operations:

"operation": "request.city"
"operation": "$uppercase($trim(request.city))"
"operation": "request.age >= 18 ? 'adult' : 'minor'"
"operation": "$join(request.addressLines, ', ')"
"operation": "{'fullName': request.firstName request.lastName, 'age': request.age}"
"operation": "request.price * 1.1"
      

Get Operator

Purpose: Perform HTTP GET requests to external APIs and store the response in the global context.

Use Cases: Fetch data from REST APIs, retrieve configuration, call external services.

Automatic Bearer Token Injection: The Multipass service automatically injects the bearer token from the Authorization header into your input data as request.bearer_token.

JSON Example:

{
  "id": "get_user_data",
  "type": "Get",
  "description": "Fetch user profile from API",
  "url": "https://api.example.com/users",
  "urlParams": [
    {"key": "id", "value": "request.userId"},
    {"key": "include", "value": "'profile,settings'"}
  ],
  "headers": "{'Authorization': 'Bearer ' request.token, 'Accept': 'application/json'}",
  "requiredElements": "data.profile",
  "key": "user_profile",
  "nextNode": "process_user"
}
      

Post Operator

Purpose: Perform HTTP POST requests to external APIs with a request body and store the response.

Use Cases: Submit data to APIs, create or update resources, authenticate, send data for processing.

JSON Example:

{
  "id": "create_user",
  "type": "Post",
  "description": "Create a new user account",
  "url": "https://api.example.com/users",
  "body": "{'email': request.email, 'name': request.name, 'age': request.age}",
  "headers": "{'Content-Type': 'application/json', 'Authorization': 'Bearer 'auth_token}",
  "requiredElements": "data.userId",
  "key": "created_user",
  "nextNode": "return_success"
}
      

Predicate Operator

Purpose: Implement conditional branching logic based on boolean expressions.

Use Cases: Validate input data, route execution, implement business rules, error handling.

JSON Example:

{
  "id": "check_age",
  "type": "Predicate",
  "description": "Check if user is an adult",
  "condition": "request.age >= 18",
  "successNode": "adult_flow",
  "failureNode": "minor_flow"
}
      

Return Operator

Purpose: Terminate the flow successfully and return data to the caller.

Use Cases: Return processed results, success responses, end flow execution.

JSON Example:

{
  "id": "return_result",
  "type": "Return",
  "description": "Return the processed user data",
  "resultExpression": "processed_user"
}
      

Fail Operator

Purpose: Terminate the flow with an error state.

Use Cases: Handle validation failures, report errors, terminate on unrecoverable conditions.

JSON Example:

{
  "id": "validation_failed",
  "type": "Fail",
  "description": "Input validation failed",
  "errorCode": 400,
  "status": "VALIDATION_ERROR",
  "errorMessage": "Required field 'email' is missing"
}
      

Error Handling in Flows

Understanding how errors propagate through flows and implementing proper error handling is crucial for building robust workflows.

Error Handling Patterns: See the original guide for detailed JSON examples of error handling, fallback, retry logic, validation chains, and error enrichment.

JSONata Expressions

JSONata is a powerful query and transformation language for JSON data. It's used extensively in Multipass flows.

Accessing Data in Expressions: Use request.field for input data, metadata.field for metadata, and $ for the entire global context.

Expression Validation: Always validate your JSONata expressions before deploying flows. Use the JSONata Exerciser to test expressions with sample data.

Complete Flow Examples

See the original guide for detailed JSON examples of flows for data transformation, API integration, multi-step orchestration, error handling, and metadata usage.

Best Practices

Flow Design: Use descriptive IDs, add descriptions, keep flows focused, plan error handling, avoid circular references.

JSONata Expressions: Test expressions, use readable code, handle nulls, avoid complexity.

Error Handling: Validate input early, provide meaningful error messages, use appropriate codes, document failure scenarios.

Metadata Usage: Use for configuration, tracking, feature flags; keep structure simple; check for existence; do not store sensitive data.

API Integration: Store tokens, use requiredElements, set headers, handle errors.

Batch Processing: Use metadata for batch-level config, design flows to be stateless, do not access other records' results.

Performance: Minimize HTTP calls, reuse computed values, cache data when possible.

Maintenance: Version flows, document logic, test with various inputs, keep under version control.

Naming Conventions: Use verb_noun for operator IDs, descriptive nouns for keys, kebab-case for flow IDs.

Testing: Test all branches, error paths, with/without metadata, batch processing.

API Usage Summary

Creating a Flow: POST /v1/multipass/flows/{flowId}

Retrieving a Flow: GET /v1/multipass/flows/{flowId}

Executing a Flow: POST /v1/multipass/flows/{flowId}/execute

Updating a Flow: PUT /v1/multipass/flows/{flowId}

Deleting a Flow: DELETE /v1/multipass/flows/{flowId}

Listing All Flows: GET /v1/multipass/flows

Key Points: Authentication required, metadata is optional, batch processing supported, request tracking via X-Request-ID, always use application/json, flows are cached, API returns appropriate status codes.

Quick Reference Card

Additional Resources