Migration Reference

This guide describes how to migrate your Acceldata Data Observability Cloud (ADOC) Python code from the legacy package acceldata-sdk (import namespace acceldata_sdk, client TorchClient) to acceldata-sdk-python (import namespace acceldata, client AdocClient).

Legacy package notice

acceldata-sdk-python is the supported SDK for catalog, pipeline, policy, and tagging workflows in ADOC. acceldata-sdk is deprecated since ADOC v26.10.0.

acceldata-sdk-python uses OpenAPI-generated models (Pydantic v2), typed resource wrappers, and a clearer separation between API transport errors and SDK workflow errors.

At a glance

Prerequisites

• Upgrade Python to 3.10 or newer. acceldata-sdk-python doesn't support Python 3.7–3.9.
• Align SDK and ADOC versions. Install an acceldata-sdk-python release that matches your ADOC deployment.
• Plan a dependency swap, not a side-by-side install. Both packages target overlapping functionality but use different import paths. Migrate imports and remove acceldata-sdk from requirements.txt or pyproject.toml when done.

Step 1: Change the package

Before:

pip install acceldata-sdk

After:

pip uninstall acceldata-sdk  # when you are ready to cut overpip install acceldata-sdk-python

Update dependency files:
- acceldata-sdk>=26.4.0+ acceldata-sdk-python>=<target-version>

Step 2: Update imports and the client

Client

Before:

from acceldata_sdk.torch_client import TorchClient
client = TorchClient(
    url="https://<your-adoc-url>",
    access_key="<access-key>",
    secret_key="<secret-key>",
    torch_connection_timeout_ms=10_000,
    torch_read_timeout_ms=20_000,
)

After:

from acceldata.client.adoc_client import AdocClient
client = AdocClient(
    url="https://<your-adoc-url>",
    access_key="<access-key>",
    secret_key="<secret-key>",
    connection_timeout_ms=10_000,
    read_timeout_ms=20_000,
)

# Optional: verify_ssl=False, or a path to a CA bundle for private TLS)

Constants

Environment variables

If you configure the client with environment variables, for example in Airflow or CI pipelines, rename the timeout variables:

URL, ACCESS_KEY, and SECRET_KEY are unchanged.

Step 3: Update error handling

Before:

from acceldata_sdk.errors import APIError, TorchSdkException

After:

from acceldata.exceptions import APIError, ApiException, AcceldataSdkException

Step 4: Update pipeline code

The largest behavioral change is the resource wrapper pattern. Legacy TorchClient methods returned Pipeline / PipelineRun objects that carried a hidden client reference and exposed methods like create_job and create_span directly on the run.

The new SDK returns PipelineResource and PipelineRunResource wrappers around generated API models. Chaining is similar, but types and some method names differ.

Create a pipeline

Before:

from acceldata_sdk.models.pipeline import CreatePipeline, PipelineMetadata
pipeline = client.create_pipeline(
    CreatePipeline(
        uid="my_pipeline",
        name="My pipeline",
        meta=PipelineMetadata(owner="team-a", team="data", codeLocation="..."),
    )
)

After:

from acceldata.models.api.pipeline.meta import Meta
from acceldata.models.sdk.pipeline.create_pipeline_input_request import CreatePipelineInputRequest
pipeline = client.create_pipeline(
    CreatePipelineInputRequest(
        uid="my_pipeline",
        name="My pipeline",
        meta=Meta(owner="team-a", team="data", code_location="..."),
    )
)
# Inspect API-shaped data:
print(pipeline.to_dict())

CreatePipelineInputRequest accepts both Meta with code_location and legacy-style objects that expose codeLocation; the SDK normalizes either form.

Load and list pipelines

Pipeline runs

Continuation IDs work the same way as in acceldata-sdk. For more information, see Pipeline Continuation IDs (Multi-Script Runs).

Jobs and lineage references

Before:

from acceldata_sdk.models.job import CreateJob, Node, JobMetadata
run.create_job(
    CreateJob(
        uid="extract_job",
        name="Extract",
        inputs=[Node(asset_uid="WAREHOUSE.db.schema.table")],
        outputs=[Node(job_uid="transform_job")],
        meta=JobMetadata(owner="etl", team="data", codeLocation="..."),
        bounded_by_span=True,
        span_uid="extract_span",
    )
)

After:

from acceldata.models.api.pipeline.meta import Meta
from acceldata.models.sdk.pipeline import CreateJobInput, JobInputOutputRef
run.create_job(
    CreateJobInput(
        uid="extract_job",
        name="Extract",
        inputs=[JobInputOutputRef(asset_uid="WAREHOUSE.db.schema.table")],
        outputs=[JobInputOutputRef(job_uid="transform_job")],
        meta=Meta(owner="etl", team="data", code_location="..."),
        bounded_by_span=True,
        span_uid="extract_span",
    )
)

Spans and events

Step 5: Update catalog code

Data sources

AssetSourceType moved to acceldata.models.sdk.catalog.asset_source_type.

Assets

Step 6: Update policy code

Most policy flows map one-to-one, but parameter names and execution scoping changed.

Get and list policies

Import paths:
# Legacy
from acceldata_sdk.constants import PolicyType, FailureStrategy, RuleExecutionStatus
from acceldata_sdk.models.ruleExecutionResult import PolicyFilter, RuleType
# New
from acceldata.models.sdk.catalog import PolicyType, PolicyFilter, RuleType, RuleExecutionStatus
from acceldata.models.sdk.catalog.executor import FailureStrategy

RuleType in the new SDK uses enum member names (for example, RuleType.DATA_QUALITY) for query parameters; acceldata-sdk used API string values such as 'DATA-QUALITY'. PolicyType API values (DATA-QUALITY, RECONCILIATION, DATA_CADENCE) are unchanged for get_policy.

Run a policy

Before (full run, synchronous):
from acceldata_sdk.constants import PolicyType, FailureStrategy
result = client.execute_policy(
    PolicyType.DATA_QUALITY,
    policy_id=123,
    sync=True,
    incremental=False,
    failure_strategy=FailureStrategy.FailOnError,
    pipeline_run_id=run_id,
)

After (equivalent):
from acceldata.models.sdk.catalog import PolicyExecutionType, RuleType
from acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInput
from acceldata.models.sdk.catalog.executor import FailureStrategy
executor = client.execute_policy(
    RuleType.DATA_QUALITY,
    123,
    PolicyExecutionInput(executionType=PolicyExecutionType.FULL),
    sync=True,
    failure_strategy=FailureStrategy.FailOnError,
    pipeline_run_id=run_id,
)
# When sync=True, executor already holds the terminal result; you can also call:
# result = executor.get_result()

Asynchronous execution:

executor = client.execute_policy(
    RuleType.DATA_QUALITY,
    123,
    PolicyExecutionInput(executionType=PolicyExecutionType.FULL),
    sync=False,
)
status = executor.get_status()
result = executor.get_result(failure_strategy=FailureStrategy.DoNotFail)

Status, results, and per-type helpers

Step 7: Update pipeline tags

Pipeline tags now use the Tag model:

from acceldata.models.api.pipeline.tag import Tag
Tag(name="env:prod", displayName="Environment: Production")

Methods not available on AdocClient

The following legacy TorchClient methods are not on AdocClient. If your code uses any of these methods, update it as described in the Notes column.

What's next

After you complete this section, explore:

• Pipelines Guide – Learn how to create pipelines, runs, jobs, and spans with acceldata-sdk-python.
• acceldata-sdk-python Overview – Review installation, client setup, and error handling in the new SDK.