Data Sources and Assets Guide

This guide describes how to use AdocClient in acceldata-sdk-python to work with data sources and assets in Acceldata Data Observability Cloud (ADOC): load data sources, run crawlers, get assets, manage metadata and sample data, and run full, incremental, or selective profiling.

ADOC supports many data source types, including warehouses, lakes, and messaging systems. In code, use AssetSourceType to filter or interpret catalog responses.

Load data sources

Load a data source by name or numeric ID. To include extended properties on the returned object, pass properties=True.

from acceldata.client.adoc_client import AdocClient
from acceldata.models.sdk.catalog import AssetSourceType
client = AdocClient(
    url="https://<your-adoc-url>",
    access_key="<your-access-key>",
    secret_key="<your-secret-key>",
)
# By name, optional properties payload
ds = client.get_datasource("snowflake_prod")
ds = client.get_datasource("snowflake_prod", properties=True)
ds = client.get_datasource(5) # same as get_datasource_by_id(5)
# All datasources
all_sources = client.get_all_datasources() # alias: get_datasources() with no filter
# Filter by source type
snowflake_only = client.get_datasources(AssetSourceType.SNOWFLAKE)

Run crawlers

You can start a crawler and check its status from the data source object (recommended) or by name on the client. Both methods support an optional RetryConfig for transient HTTP or transport issues.

# Using a DatasourceResource
ds = client.get_datasource("my_datasource")
start = ds.start_crawler()
status = ds.get_crawler_status()

# Or by name on the client
client.start_crawler("my_datasource")
client.get_crawler_status("my_datasource")

Get assets and asset types

Get an asset

To get an asset, pass a string UID (for example, Feature_bag_datasource.feature_1) or a numeric ID. The SDK treats a string of digits as a numeric ID, not a UID.

# By numeric id
asset = client.get_asset(1)

# By uid
asset = client.get_asset("Feature_bag_datasource.feature_1")

The result is an AssetResource.

List asset type definitions

types = client.get_all_asset_types()  # alias of get_asset_types()

Each entry is a catalog AssetType (name, ID, and related fields as returned by the API).

Manage metadata and sample data

from acceldata.models.sdk.catalog import CustomAssetMetadata
asset = client.get_asset("my_datasource.my_table")
# Fetched metadata (as stored for the asset)
md = asset.get_metadata()
# Or: client.get_asset_metadata(asset_id)
# Merge custom key/value pairs (string values; merged then POSTed)
asset.add_custom_metadata(
    [
        CustomAssetMetadata("owner", "analytics"),
        CustomAssetMetadata("pii", "true"),
    ]
)
# Or pass a flat mapping from the client: client.add_custom_metadata(asset_id, {"owner": "analytics"})
# Sample data (default): trigger async job and block by polling until not IN_PROGRESS
sample = asset.sample_data(sync=True) # sync=True is the default
# Or: client.sample_data(asset_id, sync=True)
# Non-blocking mode: trigger async job and return immediately with request_id
initial = asset.sample_data(sync=False)
# Or: initial = client.sample_data(asset_id, sync=False)
request_id = initial.request_id if initial is not None else None
# With an async `request_id`, poll the result
if request_id:
    result = client.get_sample_data_result(request_id)

Profiling parameters and modes

Use AssetProfilingParams to set the profiling type and an optional markerConfig (other start-profile fields use catalog defaults inside to_start_profiling_request()).

ExecutionType (from the same profiling module as ProfilingType) mirrors the same string values as ProfilingType, for execution-oriented typing in application code.

To start a profiling run, use AdocClient.profile_asset, which returns a ProfileRequestResource with get_status() and cancel(). Pass an optional transient_retry=RetryConfig(...) to retry transient HTTP or transport failures; default retry status codes include HTTP 409 (for example, when a profile is already running).

• profile_asset(..., sync=False) (default) starts profiling and immediately returns the initial request, including its ID.
• profile_asset(..., sync=True) checks the profile status until profiling reaches a terminal state.

from acceldata.client.transient_retry import RetryConfig
from acceldata.models.sdk.catalog import AssetProfilingParams, ProfilingType
params = AssetProfilingParams(profiling_type=ProfilingType.FULL)
# Async fire-and-forget (default)
profile = client.profile_asset(
    123,
    params,
    sync=False,
    # transient_retry=RetryConfig(max_attempts=6, initial_interval_seconds=30.0, max_interval_seconds=30.0),
)
# Sync mode: block until terminal status by polling with request id
completed_profile = client.profile_asset(
    123,
    params,
    sync=True,
)

AssetResource.start_profile(profiling_params=..., sync=...) returns the same ProfileRequestResource shape, with status and cancel helpers:

from acceldata.models.sdk.catalog import AssetProfilingParams, ProfilingType
res = asset.start_profile(
    AssetProfilingParams(profiling_type=ProfilingType.FULL),
    sync=False, # default: async fire-and-forget
)
print(res.to_dict())
details = res.get_status()
res.cancel()

Check profile status and cancel requests

To cancel a profile request when you have only its request ID, use cancel_profile():
client.cancel_profile(profile_req_id)

ProfileRequestResource delegates cancel and status calls to these client and service methods. You can configure transient read retries for crawler and profile calls with RetryConfig wherever the method allows it.

# Latest status for the asset — response type is MiniProfileRequest
latest = client.get_latest_profile_status(asset_id=123)
# Or: asset.get_latest_profile_status() on an AssetResource

# Specific profile request by (asset_id, req_id) — also exposed as get_profile_status
by_req = client.get_profile_request_details(asset_id=123, req_id=456)
# Or: client.get_profile_status(123, 456)

Run full profiling

Full profiling runs the configured full scan of the asset. The following example uses the client created in Load data sources.

from acceldata.models.sdk.catalog import AssetProfilingParams, ProfilingType
asset = client.get_asset("my_datasource.my_table")
params = AssetProfilingParams(profiling_type=ProfilingType.FULL)
client.profile_asset(asset.id, params)

Run incremental profiling

Use incremental profiling when the asset has an incremental strategy configured. The request must include a marker.

• For an ID-column incremental strategy, use id_marker_config.
• For other strategies, use the matching *_marker_config helper in acceldata.services.marker_config (for example, date_time_marker_config for DateTimeMarkerConfig, or file_marker_config for FileMarkerConfig — the same naming pattern as BoundsDateTimeMarkerConfig → bounds_date_time_marker_config).
• You can also set the marker type yourself by using marker_type_for or API_TYPE_BY_MARKER_CLASS, or call as_marker_config, which can fix a wrong type on a typed model instance.

AssetProfilingParams.marker_config also accepts a generated branch instance or a MarkerConfig wrapper; the client normalizes them when it sends the request.

The snippet assumes a configured AdocClient client from the start of this guide. As with full profiling, profile_asset accepts sync=True to poll until a terminal status, and transient_retry=RetryConfig(...) for transient HTTP or transport retries (including while polling in sync mode).

from acceldata.client.transient_retry import RetryConfig
from acceldata.models.sdk.catalog import AssetProfilingParams, ProfilingType
from acceldata.services.marker_config import id_marker_config
asset = client.get_asset("my_datasource.my_table")
params = AssetProfilingParams(
    profiling_type=ProfilingType.INCREMENTAL,
    marker_config=id_marker_config(
        id_column_name="pk_col",
        initial_offset=0,
    ),
)
client.profile_asset(
    asset.id,
    params,
    sync=False, # default: async; use sync=True to block until terminal
    # transient_retry=RetryConfig(max_attempts=6, initial_interval_seconds=30.0, max_interval_seconds=30.0),
)

Run selective profiling

Selective profiling runs on a bounded slice of the asset. The request must include a marker.

• For ID-bounded slices, use bounds_id_marker_config.
• For date- or file-event–bounded slices, use bounds_date_time_marker_config or bounds_file_event_marker_config.

Each example below assumes a configured client from the start of this guide. The same sync and transient_retry options described earlier apply to profile_asset for selective runs.

ID-bounded selective profiling

from acceldata.client.transient_retry import RetryConfig

from acceldata.models.sdk.catalog import AssetProfilingParams, ProfilingType

from acceldata.services.marker_config import bounds_id_marker_config
asset = client.get_asset("my_datasource.my_table")
params = AssetProfilingParams(
    profiling_type=ProfilingType.SELECTIVE,
    marker_config=bounds_id_marker_config(
        id_column_name="ID",
        from_id=0,
        to_id=1000,
    ),
)
client.profile_asset(
    asset.id,
    params,
    sync=False,
# transient_retry=RetryConfig(max_attempts=6, initial_interval_seconds=30.0, max_interval_seconds=30.0),
)

Date and time–bounded selective profiling

from acceldata.client.transient_retry import RetryConfig
from acceldata.models.sdk.catalog import AssetProfilingParams, ProfilingType
from acceldata.services.marker_config import bounds_date_time_marker_config
asset = client.get_asset("my_datasource.my_table")
params = AssetProfilingParams(
    profiling_type=ProfilingType.SELECTIVE,
    marker_config=bounds_date_time_marker_config(
        date_column_name="TO_DATE",
        format="yyyy-MM-dd",
        time_zone_id="Asia/Calcutta",
        from_date="2023-07-01 00:00:00.000",
        to_date="2024-07-14 23:59:59.999",
    ),
)
client.profile_asset(
    asset.id,
    params,
    sync=False,
    # transient_retry=RetryConfig(max_attempts=6, initial_interval_seconds=30.0, max_interval_seconds=30.0),
)

File event–bounded selective profiling

For partitioning, file-based markers, Pub/Sub, offsets, and similar cases, use the matching *_marker_config helper in acceldata.services.marker_config (each pairs with a generated *MarkerConfig class — see API_TYPE_BY_MARKER_CLASS), or build the model with an explicit type=, then pass the instance the same way as in the examples above.

from acceldata.client.transient_retry import RetryConfig
from acceldata.models.sdk.catalog import AssetProfilingParams, ProfilingType
from acceldata.services.marker_config import bounds_file_event_marker_config
asset = client.get_asset("my_datasource.my_table")
params = AssetProfilingParams(
    profiling_type=ProfilingType.SELECTIVE,
    marker_config=bounds_file_event_marker_config(
        date_column_name="ingest_date",
        time_zone_id="Asia/Calcutta",
        from_date="2019-04-01 00:00:00.000",
        to_date="2024-07-16 23:59:59.999",
    ),
)
client.profile_asset(
    asset.id,
    params,
    sync=False,
    # transient_retry=RetryConfig(max_attempts=6, initial_interval_seconds=30.0, max_interval_seconds=30.0),
)

What's next

After you complete this section, explore:

• Policy Guide – Learn how to fetch, execute, and monitor data quality and reconciliation policies against these assets.
• Unified Tags – Learn how to create tag keys and values and associate them with assets, policies, and other catalog entities.
• Pipelines Guide – Learn how job inputs and outputs reference the assets described in this guide.