Acceldata
 ADOC
Search documentation… ⌘K

Last updated: Oct 14, 2026 12:14 UTC

Policy Guide

This guide describes how to use acceldata-sdk-python to work with data quality, reconciliation, and data cadence (freshness) policies in Acceldata Data Observability Cloud (ADOC). You can get and list policies, list past executions, start full, incremental, or selective runs, check status and results, and cancel runs. In the SDK, policies are also called rules.

For information about hard-linked and soft-linked policies, see Pipeline Run Details.

Get policies and list executions

from acceldata.client.adoc_client import AdocClient
from acceldata.models.sdk.catalog import PolicyType, PolicyFilter, RuleType
client = AdocClient(
    url="https://<your-adoc-url>",
    access_key="<your-access-key>",
    secret_key="<your-secret-key>",
)
# Typed fetch (data-quality, reconciliation, or data-cadence)
rule = client.get_policy(PolicyType.RECONCILIATION, "auth001_reconciliation")
# Or untyped fetch by identifier → QualityRule (use rule_untyped.id for executions)
rule_untyped = client.get_policy(identifier="my_rule_name")
# List past executions by numeric rule id or by rule name
executions = client.policy_executions(1114, RuleType.DATA_QUALITY)
executions = client.policy_executions("dq-scala", RuleType.DATA_QUALITY)
# List executions for a rule already loaded: use the catalog rule id + RuleType, or a typed resource
recon_rule = client.get_policy(PolicyType.RECONCILIATION, "auth001_reconciliation")
recon_executions = client.policy_executions(recon_rule.rule.id, RuleType.RECONCILIATION)
recon_res = client.get_recon_policy_resource("auth001_reconciliation")
recon_executions_via_resource = recon_res.policy_executions(page=0, size=25)
# List rules with a filter
policy_filter = PolicyFilter(policy_type=RuleType.RECONCILIATION, enable=True)
listed = client.list_all_policies(policy_filter=policy_filter, page=0, size=25)

You can also get a typed resource for a single policy (data quality, reconciliation, or data cadence):
dq_res = client.get_dq_policy_resource("my_dq")
recon_res = client.get_recon_policy_resource("my_recon")
cadence_res = client.get_cadence_policy_resource("my_cadence")
# Example: list executions for that rule
dq_res.policy_executions(page=0, size=25)

Choose an execution type

Pass PolicyExecutionInput as the run payload on execute_policy, execute_dq_rule, execute_reconciliation_rule, or execute_freshness_rule. Set PolicyExecutionType to choose how data is scoped:

Type

Role

FULL

Scans the configured assets without per-run slice bounds (aside from optional Spark or rule-item options in the request).

INCREMENTAL

Runs on data since the last persisted execution marker for that rule and its assets, as defined in the catalog. Many policies need only PolicyExecutionInput(PolicyExecutionType.INCREMENTAL) with no markerConfigs; the service uses stored marker state. If your deployment requires an explicit ID-column marker for incremental runs, build it with id_marker_config in acceldata.services.marker_config, then use the same MarkerConfig / PolicyExecutionMarkerConfig pattern shown below.

SELECTIVE

Runs explicit slices. markerConfigs is required — one entry per asset (or slice) with defined slice bounds.

The sections below show markerConfigs, optional Spark settings, and related fields where they apply.

Selective runs require one marker configuration per asset slice: the catalog asset ID, plus how to bound the data (numeric ID range, date/time window, file events, Kafka-style timestamps, and other marker types the catalog supports).

  • For ID-bounded slices, use bounds_id_marker_config.
  • For other cases, use the matching *_marker_config helper in acceldata.services.marker_config that pairs with the generated *MarkerConfig class (names align — for example, BoundsDateTimeMarkerConfig → bounds_date_time_marker_config), or set type= with marker_type_for (see API_TYPE_BY_MARKER_CLASS).

Wrap the result in MarkerConfig(...), and pass a List[PolicyExecutionMarkerConfig] as markerConfigs on the request.

Run a policy with execute_policy

acceldata-sdk-python provides AdocClient.execute_policy, which runs a policy synchronously or asynchronously. The method returns an Executor instance, on which you can call get_result and get_status to obtain the execution's result and status.

get_result() is always synchronous (it is not Python async/await): the calling thread blocks in a poll loop until the run reaches a terminal status, until total_retries non-terminal poll rounds are exhausted (default -1, unlimited), or until repeated result fetches fail (optional transient retries apply per fetch). sleep_interval (default 5 seconds between polls) and failure_strategy apply to Executor.get_result / get_execution_result. FailureStrategy.DoNotFail only controls whether the client raises on terminal WARNING or ERROR outcomes; it does not make get_result non-blocking.

execute_policy parameters

The method takes the following parameters:

Parameter

Description

sync

A Boolean parameter that determines whether the policy runs synchronously or asynchronously. It is keyword-only and optional, with default False: if you omit it, the client behaves as if sync=False. If set to True, execute_policy returns only after the execution completes (the client waits for a terminal result during the execute_policy call when supported). If False, it returns as soon as the run is accepted and an execution ID is available; use get_result on the returned Executor to wait for completion.

rule_type

The policy type, specified as an enum parameter. Required. In acceldata-sdk-python, the parameter name is rule_type, of type RuleType. It accepts constant values such as RuleType.DATA_QUALITY or RuleType.RECONCILIATION. For data cadence (freshness) runs, use RuleType.DATA_CADENCE.

rule_id

The policy ID to execute, specified as a string or int parameter. Required (catalog rule ID or name, as the service accepts for that rule kind).

policy_execution_request

The run payload (PolicyExecutionInput). Required.

incremental

In other Acceldata Python clients, this may appear as a Boolean specifying whether policy execution is incremental or full, often defaulting to False. acceldata-sdk-python does not take a separate incremental argument on execute_policy. Incremental vs. full vs. selective scope is set on the request object using PolicyExecutionType: use PolicyExecutionType.INCREMENTAL for incremental execution, PolicyExecutionType.FULL for a full run, and PolicyExecutionType.SELECTIVE when supplying markerConfigs. Choosing FULL corresponds to a non-incremental full run unless another type is chosen.

pipeline_run_id

The run ID of the pipeline run where the policy is executing, specified as an optional int parameter (keyword-only). When set, it is sent as pipelineRunId on the execution request and overrides any value already set on policy_execution_request. This links the policy execution to a specific pipeline run. Alternatively, set pipelineRunId on PolicyExecutionInput.

failure_strategy

An enum parameter that determines behavior when a failure occurs. Default: DoNotFail.

failure_strategy takes an enum of type FailureStrategy, with three possible values:

  • DoNotFail: Terminal outcomes (including WARNING and ERROR result statuses) return an ExecutionResult without raising; HTTP or transport failures while polling yield a synthetic errored ExecutionResult. The same applies if total_retries is exhausted while the run is still non-terminal.
  • FailOnError: Raises PolicyExecutionCompletedWithErrorResultError when the result status indicates a failure other than WARNING. Also raises AcceldataSdkException on HTTP or transport failures during polling when building that synthetic result. A completed run whose result status is WARNING does not raise under FailOnError alone.
  • FailOnWarning: Raises PolicyExecutionCompletedWithWarningError when the result status is WARNING. It does not by itself raise on ERRORED or other error result statuses; use FailOnError for exceptions on failed (non-warning) completions.

The execution result is available through get_policy_execution_result on AdocClient, or get_result on the Executor instance returned from execute_policy; both return an ExecutionResult (or equivalent result object) when the run finishes.

Additional keyword-only parameters on execute_policy (not in the table above, but supported by this SDK):

Parameter

Description

sleep_interval

Seconds between status polls when the client waits inside execute_policy (when sync=True), or when get_result / get_execution_result is called on the handle later. Default: 5.

total_retries

Cap on non-terminal poll rounds when waiting. Default -1 (unlimited): polling continues until the run completes or result fetches fail. Set to 1 or greater to stop after that many RUNNING/WAITING poll rounds and return a synthetic errored result.

transient_retry

Optional RetryConfig for transient HTTP errors while starting the run and while polling.

get_policy_execution_result parameters

The method takes the following parameters:

Parameter

Description

rule_type

The policy type, specified as an enum parameter. Required. Accepts constant values such as RuleType.DATA_QUALITY or RuleType.RECONCILIATION (and other supported execution kinds, such as RuleType.DATA_CADENCE, where applicable).

execution_id

The execution ID to query for the result, specified as a string or int parameter. Required.

failure_strategy

An enum parameter that determines behavior when a failure occurs. Default: DoNotFail.

get_policy_execution_result also accepts an optional transient_retry for transient HTTP failures during polling. It waits until the run completes or the fetch fails — there is no poll-round cap on this method.

Check execution status

To get the current status, call get_policy_execution_status on AdocClient, or get_status on the Executor returned from execute_policy. Both report status for the same execution ID as result polling.

The required parameters for get_policy_execution_status include:

Parameter

Description

rule_type

The policy type, specified as an enum parameter. Required. Accepts constant values such as RuleType.DATA_QUALITY or RuleType.RECONCILIATION (and other supported kinds, where applicable).

execution_id

The execution ID to query for status, specified as a string or int parameter. Required.

Note

get_status  on the Executor does not take any parameters; it uses the execution ID from the run that execute_policy (or Executor.execute) already started.

Example: Run a policy asynchronously

The snippet assumes a configured AdocClient client, as shown earlier in this guide.

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
policy_execution_request = PolicyExecutionInput(PolicyExecutionType.FULL)
async_executor = client.execute_policy(
    RuleType.DATA_QUALITY,
    46,
    policy_execution_request,
    sync=False,
    failure_strategy=FailureStrategy.DoNotFail,
)
# Wait for execution to get the final result
execution_result = async_executor.get_result(failure_strategy=FailureStrategy.DoNotFail)
# Get the current status
execution_status = async_executor.get_status()

Example: Run a policy synchronously

This example also assumes a configured AdocClient client.

from acceldata.models.sdk.catalog import PolicyType, PolicyExecutionType
from acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInput
from acceldata.models.sdk.catalog.executor import FailureStrategy
policy_execution_request = PolicyExecutionInput(PolicyExecutionType.FULL)
# This will wait for execution to get the final result (within execute_policy when sync=True)
sync_executor = client.execute_policy(
    RuleType.DATA_QUALITY,
    46,
    policy_execution_request,
    sync=True,
    failure_strategy=FailureStrategy.DoNotFail,
)
# Wait for execution to get the final result 
execution_result = sync_executor.get_result(failure_strategy=FailureStrategy.DoNotFail)
# Get the current status
execution_status = sync_executor.get_status()

Get status and results by execution ID

When the execution ID is already known (for example, from another system), use get_policy_execution_status and get_policy_execution_result on AdocClient with that ID and the correct RuleType.

Method

Parameters

get_policy_execution_status(rule_type, execution_id, *, transient_retry=...)

RuleType and execution ID; optional transient retry config.

get_policy_execution_result(rule_type, execution_id, *, failure_strategy=..., transient_retry=...)

Same, plus optional FailureStrategy for when a terminal WARNING or ERROR result should raise.

Stopping a run: call cancel() on the same Executor handle. Typed policy resources may also expose a cancel_execution helper for the same purpose.

Other ways to start a run: execute_dq_rule, execute_reconciliation_rule, and execute_freshness_rule accept the same request shape; use the matching get_*_rule_result or get_policy_execution_result helpers to poll by execution ID.

Configure the execution request

Note: Execution options (markers, Spark, rule items, and so on) depend on the ADOC catalog version. Keep your acceldata-sdk-python version aligned with the server.

Pass a PolicyExecutionInput to execute_policy, execute_dq_rule, execute_reconciliation_rule, and execute_freshness_rule (the same parameter names appear in the code examples).

Use PolicyExecutionInput to:

  • Set how much data runs: FULL, INCREMENTAL, or SELECTIVE (through PolicyExecutionType).
  • For SELECTIVE, pass markerConfigs so each asset slice has bounds (see the ID, date/time, file event, and timestamp-based examples in this guide).
  • Optionally narrow which rule items run, link a pipeline run ID, tune Spark, or supply data quality dynamic SQL filter mappings.

PolicyExecutionInput fields

All fields are keyword arguments except the first.

  • executionType: Required first argument: PolicyExecutionType.FULL, INCREMENTAL, or SELECTIVE.
  • markerConfigs: Optional. List of PolicyExecutionMarkerConfig; required for SELECTIVE.
  • ruleItemSelections: Optional. Int IDs of rule items to execute; omit to run all items. In this guide, runnable examples that set this field use PolicyExecutionType.SELECTIVE (see the ID-based data quality and reconciliation examples below).
  • includeInQualityScore: Optional, default True. Includes this run in the quality score.
  • pipelineRunId: Optional. Pipeline run ID to attach.
  • sparkSQLDynamicFilterVariableMapping: Optional. For data quality policies with dynamic SQL filters.
  • sparkResourceConfig: Optional. Spark resource limits and related settings.
  • columnVariables: Optional. Column variable bindings when the policy requires them.
  • sparkFilterSelectedColumns: Optional. Column names for Spark filter selection when applicable.

Run policies by type

Use the same PolicyExecutionType and PolicyExecutionInput for data quality and reconciliation policies; only the client method changes (execute_dq_rule vs. execute_reconciliation_rule).

For data cadence (freshness), use execute_freshness_rule with the same request shape.

Loading a policy with get_policy(PolicyType.*, identifier) returns a typed DataQualityRuleResponse, ReconciliationRuleResponse, or DataCadenceRuleResponse, each with a nested rule object — use policy.rule.id in execute_*_rule (as shown in the snippets below). With get_policy(identifier=...) and no PolicyType, the client returns a QualityRule; use policy.id (there is no policy.rule on that type).

Data quality policy examples

Trigger full data quality policy: To run a full data quality policy across the entire dataset, use PolicyExecutionType.FULL on PolicyExecutionInput and call execute_dq_rule with the rule ID.

from acceldata.models.sdk.catalog import PolicyType, PolicyExecutionType
from acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInput
dq_policy = client.get_policy(PolicyType.DATA_QUALITY, "my_dq")
req = PolicyExecutionInput(PolicyExecutionType.FULL)
client.execute_dq_rule(dq_policy.rule.id, req)

Trigger incremental data quality policy: To run an incremental data quality policy based on a configured incremental strategy, use PolicyExecutionType.INCREMENTAL. The run advances from the catalog's last stored marker for that rule. The minimal request is often only the execution type; add markerConfigs only if the catalog requires an explicit marker for this policy.

dq_policy = client.get_policy(PolicyType.DATA_QUALITY, "my_dq")
req = PolicyExecutionInput(PolicyExecutionType.INCREMENTAL)
client.execute_dq_rule(dq_policy.rule.id, req)

Trigger selective data quality policy: To run a selective data quality policy over a subset of data, as determined by the chosen incremental strategy, use PolicyExecutionType.SELECTIVE with markerConfigs. Optionally, set sparkSQLDynamicFilterVariableMapping and sparkResourceConfig as shown in the ID-based example. sparkSQLDynamicFilterVariableMapping applies only when the data quality policy includes SQL filters.

ID-based selective execution

Uses a monotonically increasing column value to define data boundaries for policy execution, implemented with BoundsIdMarkerConfig. Optional Spark tuning and sparkSQLDynamicFilterVariableMapping apply when the data quality policy uses dynamic SQL filters. Optionally pass ruleItemSelections with catalog rule item definition IDs to run only those checks within the selective slice; omit the field to run every item on the policy. The same ruleItemSelections argument applies on execute_reconciliation_rule and execute_freshness_rule PolicyExecutionInput objects.

from acceldata.models.sdk.catalog import PolicyType, PolicyExecutionType
from acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInput
from acceldata.models.api.catalog.policy_execution_marker_config import PolicyExecutionMarkerConfig
from acceldata.models.api.catalog.marker_config import MarkerConfig
from acceldata.models.api.catalog.spark_resource_config import SparkResourceConfig
from acceldata.models.api.catalog.yunikorn_spark_resource_config import YunikornSparkResourceConfig
from acceldata.models.api.catalog.rule_spark_sql_dynamic_filter_variable_mapping import (
    RuleSparkSQLDynamicFilterVariableMapping,
)
from acceldata.models.api.catalog.mapping import Mapping
from acceldata.services.marker_config import bounds_id_marker_config
dq_policy = client.get_policy(PolicyType.DATA_QUALITY, "spark_sql_policy")
bounds = bounds_id_marker_config(
    id_column_name="ID",
    from_id=0,
    to_id=1000,
)
marker_configs = [
    # asset_id in the marker configuration refers to the unique identifier of the underlying asset on which the Data Quality (DQ) policy is established.
    PolicyExecutionMarkerConfig(
        asset_id=9667404,
        marker_config=MarkerConfig(bounds),
    ),
]
yunikorn = YunikornSparkResourceConfig(
    min_executors=1,
    max_executors=2,
    executor_cores=2,
    executor_memory="2g",
    driver_cores=2,
    driver_memory="2g",
)
spark_cfg = SparkResourceConfig(
    invalidFields=[],
    yunikorn=yunikorn,
    additionalConfiguration={},
)
sql_maps = [
    RuleSparkSQLDynamicFilterVariableMapping(
        rule_name="SelectiveDQPolicysparkSQLDynamicFilterVariable",
        mapping=[Mapping(key="column_name", is_column_variable=True, value="100")],
    ),
]
req = PolicyExecutionInput(
    PolicyExecutionType.SELECTIVE,
    markerConfigs=marker_configs,
    sparkResourceConfig=spark_cfg,
    sparkSQLDynamicFilterVariableMapping=sql_maps,
    ruleItemSelections=[101, 102], # replace with ids from the catalog / rule payload; omit for all items
)
client.execute_dq_rule(dq_policy.rule.id, req)

Date and time–based selective execution

Uses an increasing date column to define data boundaries for policy execution, implemented with BoundsDateTimeMarkerConfig.

from acceldata.models.sdk.catalog import PolicyType, PolicyExecutionType

from acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInput
from acceldata.models.api.catalog.policy_execution_marker_config import PolicyExecutionMarkerConfig
from acceldata.models.api.catalog.marker_config import MarkerConfig
from acceldata.services.marker_config import bounds_date_time_marker_config
dq_policy = client.get_policy(PolicyType.DATA_QUALITY, "my_dq")
bounds = bounds_date_time_marker_config(
    date_column_name="TO_DATE",
format="yyyy-MM-dd",
    from_date="2023-07-01 00:00:00.000",
    to_date="2024-07-14 23:59:59.999",
    time_zone_id="Asia/Calcutta",
)
marker_configs = [
# asset_id in the marker configuration refers to the unique identifier of the underlying asset on which the Data Quality (DQ) policy is established.
PolicyExecutionMarkerConfig(asset_id=9667404, marker_config=MarkerConfig(bounds)),
]
req = PolicyExecutionInput(PolicyExecutionType.SELECTIVE, markerConfigs=marker_configs)
client.execute_dq_rule(dq_policy.rule.id, req)

File event–based selective execution

Uses file events to establish data boundaries for policy execution, implemented with BoundsFileEventMarkerConfig. The date_column_name field must match a column on the bound asset (replace event_date with the actual column name).

from acceldata.models.sdk.catalog import PolicyType, PolicyExecutionType
from acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInput
from acceldata.models.api.catalog.policy_execution_marker_config import PolicyExecutionMarkerConfig
from acceldata.models.api.catalog.marker_config import MarkerConfig
from acceldata.services.marker_config import bounds_file_event_marker_config
dq_policy = client.get_policy(PolicyType.DATA_QUALITY, "my_dq")
bounds = bounds_file_event_marker_config(
    date_column_name="event_date",
    from_date="2024-07-01 00:00:00.000",
    to_date="2024-07-01 23:59:59.999",
    time_zone_id="Asia/Calcutta",
)
marker_configs = [
    # asset_id in the marker configuration refers to the unique identifier of the underlying asset on which the Data Quality (DQ) policy is established.
    PolicyExecutionMarkerConfig(asset_id=1202688, marker_config=MarkerConfig(bounds)),
]
req = PolicyExecutionInput(PolicyExecutionType.SELECTIVE, markerConfigs=marker_configs)
client.execute_dq_rule(dq_policy.rule.id, req)

Kafka timestamp–based selective execution

Uses offsets associated with specified timestamps to set data boundaries for policy execution, implemented with TimestampBasedMarkerConfig.

from acceldata.models.sdk.catalog import PolicyType, PolicyExecutionType
from acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInput
from acceldata.models.api.catalog.policy_execution_marker_config import PolicyExecutionMarkerConfig
from acceldata.models.api.catalog.marker_config import MarkerConfig
from acceldata.services.marker_config import timestamp_based_marker_config
dq_policy = client.get_policy(PolicyType.DATA_QUALITY, "kafka_dq_policy")
bounds = timestamp_based_marker_config(
    format="yyyy-mm-dd",
    initial_offset="2023-06-01",
    time_zone_id="Asia/Calcutta",
)
marker_configs = [
    # asset_id in the marker configuration refers to the unique identifier of the underlying asset on which the Data Quality (DQ) policy is established.
    PolicyExecutionMarkerConfig(asset_id=5241961, marker_config=MarkerConfig(bounds)),
]
req = PolicyExecutionInput(PolicyExecutionType.SELECTIVE, markerConfigs=marker_configs)
client.execute_dq_rule(dq_policy.rule.id, req)

Reconciliation policy examples

Trigger full reconciliation policy: To trigger a full reconciliation policy across the entire dataset:

from acceldata.models.sdk.catalog import PolicyType, PolicyExecutionType
from acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInput
recon_policy = client.get_policy(PolicyType.RECONCILIATION, "my_recon")
req = PolicyExecutionInput(PolicyExecutionType.FULL)
client.execute_reconciliation_rule(recon_policy.rule.id, req)

Trigger incremental reconciliation policy: To trigger an incremental reconciliation policy based on a configured incremental strategy:

recon_policy = client.get_policy(PolicyType.RECONCILIATION, "my_recon")
req = PolicyExecutionInput(PolicyExecutionType.INCREMENTAL)
client.execute_reconciliation_rule(recon_policy.rule.id, req)

Trigger selective reconciliation policy: To trigger a selective reconciliation policy over a subset of data, constrained by the selected incremental strategy:

ID-based selective execution

Uses a monotonically increasing column value to define data boundaries for policy execution, implemented with BoundsIdMarkerConfig. The asset_id in the marker configuration denotes the unique identifier of the underlying asset the reconciliation policy is based on. This can represent either the left or right asset ID.

from acceldata.models.sdk.catalog import PolicyType, PolicyExecutionType

from acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInput
from acceldata.models.api.catalog.policy_execution_marker_config import PolicyExecutionMarkerConfig
from acceldata.models.api.catalog.marker_config import MarkerConfig
from acceldata.models.api.catalog.spark_resource_config import SparkResourceConfig
from acceldata.models.api.catalog.yunikorn_spark_resource_config import YunikornSparkResourceConfig
from acceldata.services.marker_config import bounds_id_marker_config
recon_policy = client.get_policy(PolicyType.RECONCILIATION, "my_recon_policy")
bounds = bounds_id_marker_config(
    id_column_name="ID",
    from_id=0,
    to_id=1000,
)
marker_configs = [
# asset_id in the marker configuration denotes the unique identifier of the underlying asset on which the Reconciliation policy is based. This could represent either the left or right asset ID.
PolicyExecutionMarkerConfig(asset_id=9667404, marker_config=MarkerConfig(bounds)),
]
yunikorn = YunikornSparkResourceConfig(
    min_executors=1,
    max_executors=2,
    executor_cores=2,
    executor_memory="2g",
    driver_cores=2,
    driver_memory="2g",
)
spark_cfg = SparkResourceConfig(
    invalidFields=[],
    yunikorn=yunikorn,
    additionalConfiguration={},
)
req = PolicyExecutionInput(
    PolicyExecutionType.SELECTIVE,
    markerConfigs=marker_configs,
    sparkResourceConfig=spark_cfg,
    ruleItemSelections=[101, 102], # replace with ids from the catalog / rule payload; omit for all items
)
client.execute_reconciliation_rule(recon_policy.rule.id, req)

Date and time–based selective execution

Uses an increasing date column to define data boundaries for policy execution, implemented with BoundsDateTimeMarkerConfig.

from acceldata.models.sdk.catalog import PolicyType, PolicyExecutionType
from acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInput
from acceldata.models.api.catalog.policy_execution_marker_config import PolicyExecutionMarkerConfig
from acceldata.models.api.catalog.marker_config import MarkerConfig
from acceldata.services.marker_config import bounds_date_time_marker_config
recon_policy = client.get_policy(PolicyType.RECONCILIATION, "my_recon")
bounds = bounds_date_time_marker_config(
    date_column_name="TO_DATE",
    format="yyyy-MM-dd",
    from_date="2023-07-01 00:00:00.000",
    to_date="2024-07-14 23:59:59.999",
    time_zone_id="Asia/Calcutta",
)
marker_configs = [
    #asset_id in the marker configuration denotes the unique identifier of the underlying asset on which the Reconciliation policy is based. This could represent either the left or right asset ID.
    PolicyExecutionMarkerConfig(asset_id=9667404, marker_config=MarkerConfig(bounds)),
]
req = PolicyExecutionInput(PolicyExecutionType.SELECTIVE, markerConfigs=marker_configs)
client.execute_reconciliation_rule(recon_policy.rule.id, req)

File event–based selective execution

Uses file events to establish data boundaries for policy execution, implemented with BoundsFileEventMarkerConfig. Use a real date_column_name for the backing asset (same note as for data quality file-event markers).

from acceldata.models.sdk.catalog import PolicyType, PolicyExecutionType
from acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInput
from acceldata.models.api.catalog.policy_execution_marker_config import PolicyExecutionMarkerConfig
from acceldata.models.api.catalog.marker_config import MarkerConfig
from acceldata.services.marker_config import bounds_file_event_marker_config
recon_policy = client.get_policy(PolicyType.RECONCILIATION, "my_recon")
bounds = bounds_file_event_marker_config(
    date_column_name="event_date",
    from_date="2024-07-01 00:00:00.000",
    to_date="2024-07-01 23:59:59.999",
    time_zone_id="Asia/Calcutta",
)
marker_configs = [
    # asset_id in the marker configuration denotes the unique identifier of the underlying asset on which the Reconciliation policy is based. This could represent either the left or right asset ID.
    PolicyExecutionMarkerConfig(asset_id=1202688, marker_config=MarkerConfig(bounds)),
]
req = PolicyExecutionInput(PolicyExecutionType.SELECTIVE, markerConfigs=marker_configs)
client.execute_reconciliation_rule(recon_policy.rule.id, req)

Kafka timestamp–based selective execution

Uses offsets associated with specified timestamps to set data boundaries for policy execution, implemented with TimestampBasedMarkerConfig.

from acceldata.models.sdk.catalog import PolicyType, PolicyExecutionType
from acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInput
from acceldata.models.api.catalog.policy_execution_marker_config import PolicyExecutionMarkerConfig
from acceldata.models.api.catalog.marker_config import MarkerConfig
from acceldata.services.marker_config import timestamp_based_marker_config
recon_policy = client.get_policy(PolicyType.RECONCILIATION, "kafka_recon_policy")
bounds = timestamp_based_marker_config(
    format="yyyy-mm-dd",
    initial_offset="2023-06-01",
    time_zone_id="Asia/Calcutta",
)
marker_configs = [
    # asset_id in the marker configuration denotes the unique identifier of the underlying asset on which the Reconciliation policy is based. This could represent either the left or right asset ID.
    PolicyExecutionMarkerConfig(asset_id=5241961, marker_config=MarkerConfig(bounds)),
]
req = PolicyExecutionInput(PolicyExecutionType.SELECTIVE, markerConfigs=marker_configs)
client.execute_reconciliation_rule(recon_policy.rule.id, req)

What's next

After you complete this section, explore:

  • Tags and Labels: Learn how to attach tags and labels to the assets these policies run against.
  • Pipelines Guide: Learn how to link a policy execution to a pipeline run using pipelineRunId.