This guide walks through one script that demonstrates a complete pipeline lifecycle in ADOC using acceldata-sdk-python.
from acceldata.client.adoc_client import AdocClient
from acceldata.exceptions import APIError, ApiException, AcceldataSdkException
from acceldata.models.api.pipeline.meta import Meta
from acceldata.models.api.pipeline.tag import Tag
from acceldata.models.sdk.pipeline.create_job_input import CreateJobInput, JobInputOutputRef
from acceldata.models.sdk.pipeline.create_pipeline_input_request import (
CreatePipelineInputRequest,
)
from acceldata.models.sdk.pipeline.events.generic_event import GenericEvent
from acceldata.models.sdk.pipeline.events.log_event import LogEvent
from acceldata.models.sdk.pipeline.pipeline_run_input import (
PipelineRunResult,
PipelineRunStatus,
)
client = AdocClient(
url="https://<your-adoc-url>",
access_key="<your-access-key>",
secret_key="<your-secret-key>",
)
pipeline = client.create_pipeline(
CreatePipelineInputRequest(
uid="e2e_customers_pipeline",
name="E2E customers pipeline",
description="End-to-end pipeline lifecycle example",
meta=Meta(owner="data-team", team="analytics", code_location="..."),
tags=[
Tag(name="env:prod", displayName="Environment: Production"),
Tag(name="domain:finance", displayName="Domain: Finance"),
],
)
)
run = pipeline.create_pipeline_run()
result = PipelineRunResult.SUCCESS
status = PipelineRunStatus.COMPLETED
root_span = run.create_root_span(uid="e2e_customers_root")
try:
root_span.send_event(LogEvent("Run started", context_data={"run_id": run.id}))
transform_job = run.create_job(
CreateJobInput(
uid="customers_transform",
name="Transform customers",
description="End-to-end example: land curated customers for downstream analytics.",
pipeline_run_id=run.id,
bounded_by_span=True,
with_explicit_time=False,
span_uid="customers_transform_job_span",
inputs=[JobInputOutputRef(asset_uid="s3_ds.staging.customers_raw_snapshot")],
outputs=[JobInputOutputRef(asset_uid="snowflake_ds.warehouse.customers_curated")],
meta=Meta(owner="data-team", team="analytics", code_location="..."),
context={"example": "e2e_customers_pipeline"},
)
)
# Load the bounded span by UID so the example has an explicit `job_span` source.
transform_job_span = run.get_span("customers_transform_job_span")
transform_job_span.send_event(
GenericEvent(
event_uid="transform_job_started",
context_data={"job_uid": transform_job.uid},
)
)
validate_span = transform_job_span.create_child_span(
uid="customers_transform_validate_span",
associated_job_uids=[transform_job.uid],
)
validate_span.send_event(
GenericEvent(
event_uid="validate_phase_started",
context_data={"checks": "schema_and_nulls"},
)
)
validate_span.send_event(
LogEvent("Validation completed", context_data={"failed_checks": 0})
)
validate_span.end()
publish_span = transform_job_span.create_child_span(
uid="customers_transform_publish_span",
associated_job_uids=[transform_job.uid],
)
publish_span.send_event(
GenericEvent(
event_uid="publish_phase_started",
context_data={"target": "warehouse"},
)
)
publish_span.send_event(
LogEvent(
"Publish completed",
context_data={"rows_before": 120000, "rows_after": 118742},
)
)
publish_span.end()
transform_job_span.end()
root_span.send_event(
GenericEvent(
event_uid="pipeline_business_metrics",
context_data={"rows_loaded": 118742},
)
)
root_span.end(context_data={"dag_status": "SUCCESS"})
except (APIError, ApiException, AcceldataSdkException) as err:
result = PipelineRunResult.FAILURE
status = PipelineRunStatus.FAILED
root_span.failed(context_data={"error": str(err), "type": type(err).__name__})
raise
except Exception as err:
result = PipelineRunResult.FAILURE
status = PipelineRunStatus.FAILED
root_span.failed(context_data={"error": str(err), "type": "UnhandledException"})
raise
finally:
run.update_pipeline_run(
result=result,
status=status,
context_data={"example": "e2e-lifecycle"},
)
print(run.to_dict())
