POST /v1/pipelines

from metadata.sdk import configure
from metadata.sdk.entities import Pipelines
from metadata.generated.schema.api.data.createPipeline import CreatePipelineRequest

configure(
    host="https://your-company.getcollate.io/api",
    jwt_token="your-jwt-token"
)

request = CreatePipelineRequest(
    name="dbt_analytics_customers",
    displayName="DBT Customer Analytics",
    service="sample_airflow",
    description="Analytics pipeline for customer data processing",
    sourceUrl="http://localhost:8080/tree?dag_id=dbt_analytics_customers",
    tasks=[
        {
            "name": "extract_customers",
            "taskType": "BatchTask",
            "description": "Extract customer data from source",
            "downstreamTasks": ["transform_customers"]
        },
        {
            "name": "transform_customers",
            "taskType": "BatchTask",
            "description": "Transform and clean customer data",
            "downstreamTasks": ["load_customers"]
        },
        {
            "name": "load_customers",
            "taskType": "BatchTask",
            "description": "Load transformed data into warehouse"
        }
    ]
)

pipeline = Pipelines.create(request)
print(f"Created: {pipeline.fullyQualifiedName.root}")

import org.openmetadata.schema.api.data.CreatePipeline;
import org.openmetadata.schema.entity.data.Pipeline;
import org.openmetadata.sdk.fluent.Pipelines;

// client is an initialized OpenMetadataClient.
Pipelines.setDefaultClient(client);

CreatePipeline request = new CreatePipeline()
    .withName("dbt_analytics_customers")
    .withDisplayName("DBT Customer Analytics")
    .withService("sample_airflow")
    .withDescription("Analytics pipeline for customer data processing");

Pipeline pipeline = Pipelines.create(request);

curl -X POST "{base_url}/api/v1/pipelines" \
  -H "Authorization: Bearer {access_token}" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "dbt_analytics_customers",
    "displayName": "DBT Customer Analytics",
    "service": "sample_airflow",
    "description": "Analytics pipeline for customer data processing",
    "sourceUrl": "http://localhost:8080/tree?dag_id=dbt_analytics_customers",
    "tasks": [
      {
        "name": "extract_customers",
        "taskType": "BatchTask",
        "description": "Extract customer data from source",
        "downstreamTasks": ["transform_customers"]
      },
      {
        "name": "transform_customers",
        "taskType": "BatchTask",
        "description": "Transform and clean customer data",
        "downstreamTasks": ["load_customers"]
      },
      {
        "name": "load_customers",
        "taskType": "BatchTask",
        "description": "Load transformed data into warehouse"
      }
    ]
  }'

{
  "id": "538faa63-d204-46ff-aead-d158d0401cac",
  "name": "dbt_analytics_customers",
  "displayName": "DBT Customer Analytics",
  "fullyQualifiedName": "sample_airflow.dbt_analytics_customers",
  "description": "Analytics pipeline for customer data processing",
  "version": 0.1,
  "updatedAt": 1769982668397,
  "updatedBy": "admin",
  "sourceUrl": "http://localhost:8080/tree?dag_id=dbt_analytics_customers",
  "service": {
    "id": "daa58a49-df05-48a3-a417-45dfd12eacf5",
    "type": "pipelineService",
    "name": "sample_airflow",
    "fullyQualifiedName": "sample_airflow",
    "deleted": false
  },
  "serviceType": "Airflow",
  "tasks": [
    {
      "name": "extract_customers",
      "taskType": "BatchTask",
      "description": "Extract customer data from source",
      "downstreamTasks": ["transform_customers"]
    },
    {
      "name": "transform_customers",
      "taskType": "BatchTask",
      "description": "Transform and clean customer data",
      "downstreamTasks": ["load_customers"]
    },
    {
      "name": "load_customers",
      "taskType": "BatchTask",
      "description": "Load transformed data into warehouse"
    }
  ],
  "href": "http://localhost:8585/api/v1/pipelines/538faa63-d204-46ff-aead-d158d0401cac",
  "deleted": false,
  "owners": [],
  "tags": [],
  "followers": [],
  "votes": {
    "upVotes": 0,
    "downVotes": 0
  },
  "domains": []
}

POST

pipelines

POST /v1/pipelines

from metadata.sdk import configure
from metadata.sdk.entities import Pipelines
from metadata.generated.schema.api.data.createPipeline import CreatePipelineRequest

configure(
    host="https://your-company.getcollate.io/api",
    jwt_token="your-jwt-token"
)

request = CreatePipelineRequest(
    name="dbt_analytics_customers",
    displayName="DBT Customer Analytics",
    service="sample_airflow",
    description="Analytics pipeline for customer data processing",
    sourceUrl="http://localhost:8080/tree?dag_id=dbt_analytics_customers",
    tasks=[
        {
            "name": "extract_customers",
            "taskType": "BatchTask",
            "description": "Extract customer data from source",
            "downstreamTasks": ["transform_customers"]
        },
        {
            "name": "transform_customers",
            "taskType": "BatchTask",
            "description": "Transform and clean customer data",
            "downstreamTasks": ["load_customers"]
        },
        {
            "name": "load_customers",
            "taskType": "BatchTask",
            "description": "Load transformed data into warehouse"
        }
    ]
)

pipeline = Pipelines.create(request)
print(f"Created: {pipeline.fullyQualifiedName.root}")

import org.openmetadata.schema.api.data.CreatePipeline;
import org.openmetadata.schema.entity.data.Pipeline;
import org.openmetadata.sdk.fluent.Pipelines;

// client is an initialized OpenMetadataClient.
Pipelines.setDefaultClient(client);

CreatePipeline request = new CreatePipeline()
    .withName("dbt_analytics_customers")
    .withDisplayName("DBT Customer Analytics")
    .withService("sample_airflow")
    .withDescription("Analytics pipeline for customer data processing");

Pipeline pipeline = Pipelines.create(request);

curl -X POST "{base_url}/api/v1/pipelines" \
  -H "Authorization: Bearer {access_token}" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "dbt_analytics_customers",
    "displayName": "DBT Customer Analytics",
    "service": "sample_airflow",
    "description": "Analytics pipeline for customer data processing",
    "sourceUrl": "http://localhost:8080/tree?dag_id=dbt_analytics_customers",
    "tasks": [
      {
        "name": "extract_customers",
        "taskType": "BatchTask",
        "description": "Extract customer data from source",
        "downstreamTasks": ["transform_customers"]
      },
      {
        "name": "transform_customers",
        "taskType": "BatchTask",
        "description": "Transform and clean customer data",
        "downstreamTasks": ["load_customers"]
      },
      {
        "name": "load_customers",
        "taskType": "BatchTask",
        "description": "Load transformed data into warehouse"
      }
    ]
  }'

{
  "id": "538faa63-d204-46ff-aead-d158d0401cac",
  "name": "dbt_analytics_customers",
  "displayName": "DBT Customer Analytics",
  "fullyQualifiedName": "sample_airflow.dbt_analytics_customers",
  "description": "Analytics pipeline for customer data processing",
  "version": 0.1,
  "updatedAt": 1769982668397,
  "updatedBy": "admin",
  "sourceUrl": "http://localhost:8080/tree?dag_id=dbt_analytics_customers",
  "service": {
    "id": "daa58a49-df05-48a3-a417-45dfd12eacf5",
    "type": "pipelineService",
    "name": "sample_airflow",
    "fullyQualifiedName": "sample_airflow",
    "deleted": false
  },
  "serviceType": "Airflow",
  "tasks": [
    {
      "name": "extract_customers",
      "taskType": "BatchTask",
      "description": "Extract customer data from source",
      "downstreamTasks": ["transform_customers"]
    },
    {
      "name": "transform_customers",
      "taskType": "BatchTask",
      "description": "Transform and clean customer data",
      "downstreamTasks": ["load_customers"]
    },
    {
      "name": "load_customers",
      "taskType": "BatchTask",
      "description": "Load transformed data into warehouse"
    }
  ],
  "href": "http://localhost:8585/api/v1/pipelines/538faa63-d204-46ff-aead-d158d0401cac",
  "deleted": false,
  "owners": [],
  "tags": [],
  "followers": [],
  "votes": {
    "upVotes": 0,
    "downVotes": 0
  },
  "domains": []
}

Create a Pipeline

Create a new pipeline within a pipeline service.

Body Parameters

string

required

Name of the pipeline. Must be unique within the parent pipeline service.

string

required

Fully qualified name of the parent PipelineService, for example sample_airflow.

string

Human-readable display name for the pipeline.

string

Description of the pipeline in Markdown format.

string

URL to the pipeline definition in the source system, for example an Airflow DAG URL.

array

Array of pipeline tasks representing individual steps in the pipeline.

Show properties

string

Name of the task.

string

URL to display for this task.

string

URL to the task in the source system.

string

Type of the task, for example BatchTask, StreamTask.

string

Description of the task.

array

Array of task names that depend on this task.

array

Array of owner references (users or teams) to assign to the pipeline.

Show properties

string

UUID of the owner entity.

string

Type of owner entity, for example user, team.

string

Name of the owner entity.

array

Fully qualified names of the domains to assign for governance purposes.

array

Array of classification tags to apply to the pipeline.

Show properties

string

required

Fully qualified name of the tag.

string

Type of label, for example Manual, Derived, Propagated.

string

State of the tag, for example Suggested, Confirmed.

object

Custom property values defined by your organization’s metadata schema.

POST /v1/pipelines

from metadata.sdk import configure
from metadata.sdk.entities import Pipelines
from metadata.generated.schema.api.data.createPipeline import CreatePipelineRequest

configure(
    host="https://your-company.getcollate.io/api",
    jwt_token="your-jwt-token"
)

request = CreatePipelineRequest(
    name="dbt_analytics_customers",
    displayName="DBT Customer Analytics",
    service="sample_airflow",
    description="Analytics pipeline for customer data processing",
    sourceUrl="http://localhost:8080/tree?dag_id=dbt_analytics_customers",
    tasks=[
        {
            "name": "extract_customers",
            "taskType": "BatchTask",
            "description": "Extract customer data from source",
            "downstreamTasks": ["transform_customers"]
        },
        {
            "name": "transform_customers",
            "taskType": "BatchTask",
            "description": "Transform and clean customer data",
            "downstreamTasks": ["load_customers"]
        },
        {
            "name": "load_customers",
            "taskType": "BatchTask",
            "description": "Load transformed data into warehouse"
        }
    ]
)

pipeline = Pipelines.create(request)
print(f"Created: {pipeline.fullyQualifiedName.root}")

import org.openmetadata.schema.api.data.CreatePipeline;
import org.openmetadata.schema.entity.data.Pipeline;
import org.openmetadata.sdk.fluent.Pipelines;

// client is an initialized OpenMetadataClient.
Pipelines.setDefaultClient(client);

CreatePipeline request = new CreatePipeline()
    .withName("dbt_analytics_customers")
    .withDisplayName("DBT Customer Analytics")
    .withService("sample_airflow")
    .withDescription("Analytics pipeline for customer data processing");

Pipeline pipeline = Pipelines.create(request);

curl -X POST "{base_url}/api/v1/pipelines" \
  -H "Authorization: Bearer {access_token}" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "dbt_analytics_customers",
    "displayName": "DBT Customer Analytics",
    "service": "sample_airflow",
    "description": "Analytics pipeline for customer data processing",
    "sourceUrl": "http://localhost:8080/tree?dag_id=dbt_analytics_customers",
    "tasks": [
      {
        "name": "extract_customers",
        "taskType": "BatchTask",
        "description": "Extract customer data from source",
        "downstreamTasks": ["transform_customers"]
      },
      {
        "name": "transform_customers",
        "taskType": "BatchTask",
        "description": "Transform and clean customer data",
        "downstreamTasks": ["load_customers"]
      },
      {
        "name": "load_customers",
        "taskType": "BatchTask",
        "description": "Load transformed data into warehouse"
      }
    ]
  }'

{
  "id": "538faa63-d204-46ff-aead-d158d0401cac",
  "name": "dbt_analytics_customers",
  "displayName": "DBT Customer Analytics",
  "fullyQualifiedName": "sample_airflow.dbt_analytics_customers",
  "description": "Analytics pipeline for customer data processing",
  "version": 0.1,
  "updatedAt": 1769982668397,
  "updatedBy": "admin",
  "sourceUrl": "http://localhost:8080/tree?dag_id=dbt_analytics_customers",
  "service": {
    "id": "daa58a49-df05-48a3-a417-45dfd12eacf5",
    "type": "pipelineService",
    "name": "sample_airflow",
    "fullyQualifiedName": "sample_airflow",
    "deleted": false
  },
  "serviceType": "Airflow",
  "tasks": [
    {
      "name": "extract_customers",
      "taskType": "BatchTask",
      "description": "Extract customer data from source",
      "downstreamTasks": ["transform_customers"]
    },
    {
      "name": "transform_customers",
      "taskType": "BatchTask",
      "description": "Transform and clean customer data",
      "downstreamTasks": ["load_customers"]
    },
    {
      "name": "load_customers",
      "taskType": "BatchTask",
      "description": "Load transformed data into warehouse"
    }
  ],
  "href": "http://localhost:8585/api/v1/pipelines/538faa63-d204-46ff-aead-d158d0401cac",
  "deleted": false,
  "owners": [],
  "tags": [],
  "followers": [],
  "votes": {
    "upVotes": 0,
    "downVotes": 0
  },
  "domains": []
}

Returns

Returns the created pipeline object with all specified properties and system-generated fields.

Response

string

Unique identifier for the pipeline (UUID format).

string

Pipeline name.

string

Fully qualified name in format service.pipelineName.

string

Human-readable display name.

string

Description of the pipeline in Markdown format.

string

URL to the pipeline in the source system.

object

Reference to the parent pipeline service.

Show properties

string

UUID of the pipeline service.

string

Type of entity (always pipelineService).

string

Name of the pipeline service.

string

Fully qualified name of the pipeline service.

string

Type of pipeline service, for example Airflow, Dagster, DBTCloud.

array

List of tasks in the pipeline.

Show properties

string

Name of the task.

string

Type of the task.

string

Description of the task.

array

Names of downstream dependent tasks.

array

List of owners assigned to the pipeline.

Show properties

string

UUID of the owner entity.

string

Type of owner entity, for example user, team.

string

Name of the owner entity.

array

Domain assignments for governance.

array

Classification tags applied to the pipeline.

Show properties

string

Fully qualified name of the tag.

string

Type of label, for example Manual, Derived, Propagated.

string

State of the tag, for example Suggested, Confirmed.

object

Custom property values defined by your organization’s metadata schema.

number

Version number for the entity (starts at 0.1).

Create or Update (PUT)

Use PUT /v1/pipelines instead of POST to perform an upsert. If a pipeline with the same fullyQualifiedName already exists, it will be updated; otherwise, a new pipeline is created. The request body is the same as POST.

curl -X PUT "{base_url}/api/v1/pipelines" \
  -H "Authorization: Bearer {access_token}" \
  -H "Content-Type: application/json" \
  -d '{ ... same body as POST ... }'

Note: PUT will not return a 409 conflict error if the entity already exists — it will update the existing entity instead. However, for certain fields (such as description, tags, owners, and custom properties), PUT preserves an already-populated value instead of overwriting it with an empty or omitted one — the exact behavior depends on the field and the calling user or bot’s permissions, not a blanket rule. See Entity Update Behavior for the full preserve/merge rules. Use PATCH to intentionally replace an existing value.

Bulk Create or Update (PUT)

Use PUT /v1/pipelines/bulk to create or update multiple pipelines in a single request. The request body is an array of create request objects.

curl -X PUT "{base_url}/api/v1/pipelines/bulk" \
  -H "Authorization: Bearer {access_token}" \
  -H "Content-Type: application/json" \
  -d '[
    { "name": "pipeline_one", "service": "sample_airflow" },
    { "name": "pipeline_two", "service": "sample_airflow" }
  ]'

Error Handling

Code	Error Type	Description
`400`	`BAD_REQUEST`	Invalid request body or missing required fields
`401`	`UNAUTHORIZED`	Invalid or missing authentication token
`403`	`FORBIDDEN`	User lacks permission to create pipelines
`409`	`ENTITY_ALREADY_EXISTS`	Pipeline with same name already exists in service (POST only)

API Reference

SDK

API

Data Assets

AI Assets

Discovery

Lineage

Data Contracts

Teams & Users

Data Governance

Data Quality

Alerts

Incidents

Billing

Query Runner

Context Plugins

Standards

Create a Pipeline

Create a Pipeline

Body Parameters

Returns

Response

Create or Update (PUT)

Bulk Create or Update (PUT)

Error Handling