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.open-metadata.org/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;
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
POST
/
v1
/
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.open-metadata.org/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;
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
Array of pipeline tasks representing individual steps in the pipeline.
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.open-metadata.org/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;
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
Create or Update (PUT)
UsePUT /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 ... }'
PUT will not return a 409 conflict error if the entity already exists — it will update the existing entity instead.Bulk Create or Update (PUT)
UsePUT /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) |
Was this page helpful?