Acceldata
 ADOC
Search documentation… ⌘K

Last updated: Sep 29, 2026 09:19 UTC

Policy Guide

Data quality, reconciliation, and data cadence (freshness) policies in ADOC are catalog rules. This guide covers discovering policies, listing executions, starting runs (full, incremental, or selective), and reading status, results, and cancellation from Python.

Related product docs: For hard-linked vs. soft-linked policy behavior, see Pipeline Run Details in the ADOC User Guide documentation.

Get a Policy, List Policies, List Executions 

from acceldata.client.adoc_client import AdocClientfrom acceldata.models.sdk.catalog import PolicyType, PolicyFilter, RuleTypeclient = 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 nameexecutions = 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 resourcerecon_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 filterpolicy_filter = PolicyFilter(policy_type=RuleType.RECONCILIATION, enable=True)listed = client.list_all_policies(policy_filter=policy_filter, page=0, size=25)

Typed resources are also available for a single rule (data quality, reconciliation, or 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 ruledq_res.policy_executions(page=0, size=25)

Starting a Run (Full, Incremental, or Selective)

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 always need one row 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.

execute_policy (Sync and Async) and the Run Handle

acceldata-sdk-python exposes AdocClient.execute_policy, which executes policies both synchronously and 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.

Parameters for Executing Policies (execute_policy)

The required parameters include:

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.

Parameters to Retrieve Policy Execution Results

The required parameters 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 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.

Note: For more information on hard-linked and soft-linked policies, see the ADOC product documentation for rules and executions (also noted under Related product docs at the top of this guide).

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.

Status: Client Methods and Executor.get_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.

Asynchronous Execution Example

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

from acceldata.models.sdk.catalog import PolicyExecutionType, RuleTypefrom acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInputfrom acceldata.models.sdk.catalog.executor import FailureStrategypolicy_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 resultexecution_result = async_executor.get_result(failure_strategy=FailureStrategy.DoNotFail)# Get the current statusexecution_status = async_executor.get_status()

Synchronous Execution Example

Same client assumption as the asynchronous example.

from acceldata.models.sdk.catalog import PolicyType, PolicyExecutionTypefrom acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInputfrom acceldata.models.sdk.catalog.executor import FailureStrategypolicy_execution_request = PolicyExecutionInput(PolicyExecutionType.FULL)# This waits 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 resultexecution_result = sync_executor.get_result(failure_strategy=FailureStrategy.DoNotFail)# Get the current statusexecution_status = sync_executor.get_status()
cancellation = sync_executor.cancel()

Cancel Execution Example

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.

Trigger Policies: 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 in normal setups.
  • 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.

execute_dq_rule and execute_reconciliation_rule (Trigger Runs)

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 Execution 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, PolicyExecutionTypefrom acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInputdq_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, sparkSQLDynamicFilterVariableMapping and sparkResourceConfig are available, as shown in the optional example below. sparkSQLDynamicFilterVariableMapping applies only when the data quality policy includes SQL filters.

ID-Based Selective Policy 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[INLINE_CODE]PolicyExecutionInput[/INLINE_CODE] objects.

from acceldata.models.sdk.catalog import PolicyType, PolicyExecutionTypefrom acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInputfrom acceldata.models.api.catalog.policy_execution_marker_config import PolicyExecutionMarkerConfigfrom acceldata.models.api.catalog.marker_config import MarkerConfigfrom acceldata.models.api.catalog.spark_resource_config import SparkResourceConfigfrom acceldata.models.api.catalog.yunikorn_spark_resource_config import YunikornSparkResourceConfigfrom acceldata.models.api.catalog.rule_spark_sql_dynamic_filter_variable_mapping import ( RuleSparkSQLDynamicFilterVariableMapping,)from acceldata.models.api.catalog.mapping import Mappingfrom acceldata.services.marker_config import bounds_id_marker_configdq_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)

DateTime-Based Selective Policy Execution

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

from acceldata.models.sdk.catalog import PolicyType, PolicyExecutionTypefrom acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInputfrom acceldata.models.api.catalog.policy_execution_marker_config import PolicyExecutionMarkerConfigfrom acceldata.models.api.catalog.marker_config import MarkerConfigfrom acceldata.services.marker_config import bounds_date_time_marker_configdq_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 Policy 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, PolicyExecutionTypefrom acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInputfrom acceldata.models.api.catalog.policy_execution_marker_config import PolicyExecutionMarkerConfigfrom acceldata.models.api.catalog.marker_config import MarkerConfigfrom acceldata.services.marker_config import bounds_file_event_marker_configdq_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 Policy Execution

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

from acceldata.models.sdk.catalog import PolicyType, PolicyExecutionTypefrom acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInputfrom acceldata.models.api.catalog.policy_execution_marker_config import PolicyExecutionMarkerConfigfrom acceldata.models.api.catalog.marker_config import MarkerConfigfrom acceldata.services.marker_config import timestamp_based_marker_configdq_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 Execution Examples

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

from acceldata.models.sdk.catalog import PolicyType, PolicyExecutionTypefrom acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInputrecon_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 Policy 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, PolicyExecutionTypefrom acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInputfrom acceldata.models.api.catalog.policy_execution_marker_config import PolicyExecutionMarkerConfigfrom acceldata.models.api.catalog.marker_config import MarkerConfigfrom acceldata.models.api.catalog.spark_resource_config import SparkResourceConfigfrom acceldata.models.api.catalog.yunikorn_spark_resource_config import YunikornSparkResourceConfigfrom acceldata.services.marker_config import bounds_id_marker_configrecon_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)

DateTime-Based Selective Policy Execution

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

from acceldata.models.sdk.catalog import PolicyType, PolicyExecutionTypefrom acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInputfrom acceldata.models.api.catalog.policy_execution_marker_config import PolicyExecutionMarkerConfigfrom acceldata.models.api.catalog.marker_config import MarkerConfigfrom acceldata.services.marker_config import bounds_date_time_marker_configrecon_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 Policy 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, PolicyExecutionTypefrom acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInputfrom acceldata.models.api.catalog.policy_execution_marker_config import PolicyExecutionMarkerConfigfrom acceldata.models.api.catalog.marker_config import MarkerConfigfrom acceldata.services.marker_config import bounds_file_event_marker_configrecon_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 Policy Execution

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

from acceldata.models.sdk.catalog import PolicyType, PolicyExecutionTypefrom acceldata.models.sdk.catalog.policy_execution_request import PolicyExecutionInputfrom acceldata.models.api.catalog.policy_execution_marker_config import PolicyExecutionMarkerConfigfrom acceldata.models.api.catalog.marker_config import MarkerConfigfrom acceldata.services.marker_config import timestamp_based_marker_configrecon_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.