create_data_pipeline

DataPipelinesApi.create_data_pipeline(role_id, create_data_pipeline_request_body, content_type=None, _request_timeout=None, _request_auth=None, _content_type=None, _headers=None, _host_index=0)

Create a Glue or AI Flow data pipeline

Creates a data pipeline asynchronously. DataPipelineType=ai uses Bedrock Flows; omitted or glue uses Glue workflows. Glue requires DataPipelineName, Nodes, Graph, Description, and Keywords. Name is 3-50 alphanumeric/underscore (IPV-1004 / IPV-1018). Glue writes create_in_progress then starts Step Functions; poll GET /data-pipelines/{id} for PipelineStatus. AI create rejects extra body keys (IPV-1078) and may return LambdaUploadUrls for LambdaFunction nodes. Glue validate_data_pipeline_body failures return HTTP 501 DP-1010 (not 400). Empty body is GE-1001 (HTTP 403).

Parameters:

• role_id (str) – Amorphic role ID the request is authorized against. Must be a role the user belongs to. Lambda returns AUTH-1001 (HTTP 403 on v2-dataPipelines, HTTP 400 on v2-downloadLogs) if missing after gateway validation. (required)

• create_data_pipeline_request_body (CreateDataPipelineRequestBody) – (required)

• content_type (str)

• _request_timeout (int, tuple(int, int), optional) – timeout setting for this request. If one number provided, it will be total request timeout. It can also be a pair (tuple) of (connection, read) timeouts.

• _request_auth (dict, optional) – set to override the auth_settings for an a single request; this effectively ignores the authentication in the spec for a single request.

• _content_type (str, Optional) – force content-type for the request.

• _headers (dict, optional) – set to override the headers for a single request; this effectively ignores the headers in the spec for a single request.

• _host_index (int, optional) – set to override the host_index for a single request; this effectively ignores the host_index in the spec for a single request.

Return type:

CreateDataPipelineResponse

Returns:

Returns the result object.

Request and Response Examples

HTTP: POST /data-pipelines

Request

Glue ETL pipeline with start, job, and end nodes

{
  "DataPipelineName": "orders_enrichment_flow",
  "Description": "Enrich orders with customer attributes",
  "DataPipelineType": "glue",
  "Keywords": [
    "orders",
    "etl"
  ],
  "MaxConcurrentRuns": 1,
  "Nodes": [
    {
      "ModuleType": "start_node",
      "NodeName": "start"
    },
    {
      "ModuleType": "etl_job",
      "NodeName": "enrich_orders",
      "Resource": {
        "Id": "job-11111111-1111-1111-1111-111111111111",
        "Name": "enrich_orders_job"
      },
      "Arguments": {}
    },
    {
      "ModuleType": "end_node",
      "NodeName": "end"
    }
  ],
  "Graph": {
    "start": {
      "success": [
        "enrich_orders"
      ],
      "failure": []
    },
    "enrich_orders": {
      "success": [
        "end"
      ],
      "failure": []
    },
    "end": {
      "success": [],
      "failure": []
    }
  }
}

AI Flow with Input and Output nodes

{
  "DataPipelineName": "kb_qa_flow",
  "DataPipelineType": "ai",
  "Description": "Answer questions from a knowledge base",
  "Keywords": [
    "ai"
  ],
  "DefaultExecutionProperties": {
    "InputString": "Summarize last quarter orders"
  },
  "Nodes": [
    {
      "ModuleType": "Input",
      "NodeName": "input",
      "Outputs": [
        {
          "Name": "document",
          "Type": "String"
        }
      ]
    },
    {
      "ModuleType": "Output",
      "NodeName": "output",
      "Inputs": [
        {
          "Name": "document",
          "Type": "String",
          "Expression": "$.data.document"
        }
      ]
    }
  ],
  "Graph": {
    "input.document": {
      "success": [
        "output.document"
      ],
      "failure": []
    }
  }
}

Success (HTTP 200)

Creation initiated. Glue Message is “Data pipeline creation initiated.” AI may include LambdaUploadUrls.

Glue create started

{
  "DataPipelineId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "Message": "Data pipeline creation initiated."
}

AI Flow create started with Lambda upload URLs

{
  "DataPipelineId": "b2c3d4e5-f6a7-8901-bcde-f12345678901",
  "Message": "Data pipeline creation initiated.",
  "LambdaUploadUrls": {
    "custom_fn": {
      "presignedUrl": "https://s3.amazonaws.com/bucket/lambda.zip?X-Amz-Algorithm=AWS4-HMAC-SHA256",
      "s3Location": "s3://bucket/ai-flows/kb_qa_flow/custom_fn.zip"
    }
  }
}

Errors

Documented error codes: AUTH-1001, GE-1046, AUTH-1012, GE-1028, GE-1044.

HTTP 400

API Gateway request validation (missing required parameters such as role_id). Lambda InvalidInputException from this function is HTTP 403.

Gateway rejected missing required header

{
  "Message": "Invalid request"
}

HTTP 403

Invalid input or unauthorized. dataPipelines maps InvalidInputException and UnauthorizedUserException to HTTP 403.

Missing role_id after gateway

{
  "Message": "AUTH-1001 - Missing role information in event header"
}

Unknown pipeline or execution id

{
  "Message": "GE-1046 - Invalid DataPipelineId id received - 00000000-0000-0000-0000-000000000000"
}

Insufficient access on the pipeline

{
  "Message": "AUTH-1012 - User: user1 requires at least read-only access on the resource to perform this action."
}

Listing limit greater than 1000

{
  "Message": "GE-1028 - Out of range for items per page, Limit is 1000"
}

HTTP 500

Backend failure (GenericFailureException, InconsistentMetadataException, or unexpected GE-1044).

Unhandled exception

{
  "Message": "GE-1044 - Unexpected error - An unexpected error occurred."
}