Invoke Cognite Functions

Call already-deployed validation handlers in Cognite Functions from Python. Invoke helpers send payloads to the function running in CDF — they do not run validation locally.

To choose between a local sample (run_validation()) and a full estate historic scan, see Confirm that rules work.

For local iteration without deployed infrastructure, use Run validation locally. To deploy infrastructure or run the validation pipeline, see Deploy infrastructure and pipeline.

Local validation vs Cognite Function invocation

	run_validation()	Invoke helpers (call_validate_instances_shacl, call_validation, etc.)
Where it runs	Locally in your Python environment	In a deployed Cognite Function in CDF
Requires deployed function	No	Yes — deploy first with deploy_validation_infrastructure()
Typical use	Iterating on rules, quick checks, notebooks	Triggering validation jobs from scripts or orchestration

Use run_validation() when you want to validate data without Cognite Functions. Use the invoke helpers to trigger the already-deployed function running in CDF.

Quick example

from cognite_data_quality import load_cognite_client_from_toml, call_validate_instances_shacl

client = load_cognite_client_from_toml("config.toml")

data = {
    "instances": {"items": {"n": [...]}},  # From workflow trigger or your own query
    "ruleset_references": [{"externalId": "my_shacl_rules", "version": "1.0.0"}],
    "datamodel_space": "my_space",
    "datamodel_external_id": "MyDataModel",
    "datamodel_version": "v1",
    "records_config": {
        "stream_id": "dq_validation_stream",
        "rule_set_id": "MyViewSHACLv1",
        "rule_set_version": "1.0",
    },
}

result = call_validate_instances_shacl(client, data)
print(result.get("conforms"), result.get("instances_validated"))

Invoke helpers

The following helpers wrap call_validation() for specific validation types. All require a deployed function — run deploy_validation_infrastructure() first.

Use ruleset_references (Data Product + RuleSet) or inline shacl_rules in payloads. CDF Files SHACL loading (shacl_rules_file_external_id) is not supported in 0.4.0.

To deploy infrastructure or run the validation pipeline (deploy_validation_pipeline()), see Deploy infrastructure and pipeline — not the invoke helpers.

Unified entry point

from cognite_data_quality import call_validation

result = call_validation(
    client=client,
    validation_type="instance",  # or "instance_sync_cursor", "timeseries", "orchestrator", "partitioned", "test", "aggregate_uniqueness", "shacl"
    data=data,
)

Convenience wrappers

Function	Validation type	Use case
deploy_validation_pipeline()	orchestrator or enqueue	Deploy and run full pipeline (parallel orchestrator or queue enqueue)
enqueue_historic_validation()	enqueue	Enqueue historic jobs on HistoricJobQueue without parallel orchestrator
call_validate_instances_shacl()	instance	Validate instances (single invocation)
call_validate_instances_shacl_partitioned()	partitioned	Worker for partitioned validation
call_validate_timeseries_shacl()	timeseries	Validate time series data
call_validate_shacl()	shacl	Discover dqs:unique from SHACL and run aggregate uniqueness (scheduled workflow uses this)
call_validate_aggregate_uniqueness()	aggregate_uniqueness	Low-level aggregate uniqueness executor (explicit payload; no SHACL discovery)
call_validation(validation_type="instance_sync_cursor", ...)	instance_sync_cursor	Cursor-based incremental validation (sync cursor mode)

Each helper is a wrapper around call_validation() with the appropriate validation_type. Pass a payload dict and optional wait=True (default). See the function docstrings for payload keys.

job_run_id requirements

Several handlers require an explicit job_run_id in the payload (or in workflow trigger input). Silent timestamp fallbacks were removed so historic runs stay visible in dashboards.

Validation type	job_run_id
instance_sync_cursor	Required — resolved from sync cursor state or payload; must match historic run when continuing after orchestration
shacl (scheduled uniqueness)	Minted per run as job_{view}_uniqueness_{epoch} when omitted; explicit payload id is honored
aggregate_uniqueness	Required when posting records (low-level executor; does not mint)
orchestrator	Sets canonical id for partitions, sync handoff, and edge baseline
partitioned	Passed from orchestrator workflow input
instance (trigger-based)	Optional — defaults to timestamp if omitted

Canonical historic format: job_{view_external_id}_{epoch_seconds} (view external ID lowercased). Produced by deploy_validation_pipeline() and the historic queue manager so queue- and manual-enqueued runs align.

Legacy: older hq_* ids do not match new job_* records. Filter dashboards by the id on the orchestration you triggered.

Do not bake job_run_id into deploy-time sync trigger templates — the orchestrator handoff and sync cursor state supply it.

See Sync-cursor and historic orchestration and Troubleshooting.

Enqueue historic validation (0.4.4+)

This is track B in Confirm that rules work: a full estate historic against published DataProduct / RuleSet versions. It does not validate unpublished local TTL. A completed job for the same version is skipped unless you pass force=True (re-scan the same published rules; still skips queued / running).

Enqueue historic jobs on HistoricJobQueue for sequential processing by dq-historic-queue-manager. Requires prior deploy_validation_infrastructure() and DMS DataQualitySettings with managed_data_product_external_ids.

from cognite_data_quality import enqueue_historic_validation

# All managed DataProducts
result = enqueue_historic_validation(
    client,
    data_quality_space="sp_data_quality",
    trigger_queue_manager=True,
)
# result["mode"] == "enqueue"
# result: status, data_products, total_enqueued, total_skipped, warnings, queue_manager_execution_id

# Single DataProduct, optional view filter
result = enqueue_historic_validation(
    client,
    data_product_external_id="my-data-product",
    data_product_version="1.2.0",  # optional; latest published if omitted
    view_external_id="Equipment",
    view_space="sp_rmdm_dm",
    trigger_queue_manager=True,
)

# Re-run historic for all ruled views without a DataProduct/RuleSet version bump.
# trigger_queue_manager=True starts work; without it the job stays queued until cron.
result = enqueue_historic_validation(
    client,
    data_product_external_id="my-data-product",
    force=True,
    trigger_queue_manager=True,
)

Equivalent via deploy_validation_pipeline():

from cognite_data_quality import deploy_validation_pipeline

result = deploy_validation_pipeline(
    client,
    data_product_external_id="my-data-product",
    historic_mode="enqueue",
    trigger_queue_manager=True,
)

Return shape: enqueue mode returns a dict with mode: "enqueue". Default orchestrator mode returns a list of per-view trigger results (view, call_id, status, …). Do not assume a single return type across modes.

See Deploy → Historic queue enqueue mode and Migration 0.4.4.

Payload reference

Pre-persist Save gate

Interactive Save before DMS apply uses validation_type: "pre_persist" — a dedicated type, not instance or test. Overlay and hydrate contract: Pre-persist Save gate. Identify the published Data Product, RuleSet, and view — not a DMS data model.

from cognite_data_quality import call_validation

result = call_validation(
    client,
    validation_type="pre_persist",
    data={
        "job_run_id": "save-trace-001",
        "data_product": {"externalId": "people-product", "version": "1.0.0"},
        "ruleset_references": [{"externalId": "my-ruleset", "version": "1.0.0"}],
        "proposed": [
            {
                "space": "sp_people",
                "externalId": "person-1",
                "properties": {"sp_people": {"Person": {"name": "Ada"}}},
            }
        ],
        "hydrate_from_dms": {
            "views": [{"space": "sp_people", "externalId": "Person", "version": "v1"}],
            "filter": {"equals": {"property": ["sp_people", "Person", "orgVersion"], "value": "org-v3"}},
            "limit": 250,
        },
        "auto_load_depth": 2,
        "max_auto_load_instances": 250,
    },
)

post_to_records defaults false. Empty or ID-only proposed must fail. Azure Function data is 36 kB — hydrate by filter; do not ship the neighbourhood in data.

Instance validation

For instance validation (single or partitioned), the payload typically includes:

• instances: Dict with items (e.g. {"items": {"n": [...]}})
• ruleset_references or shacl_rules (inline TTL string)
• datamodel_space, datamodel_external_id, datamodel_version
• records_config: stream_id, rule_set_id, rule_set_version, optional records_space, records_container
• auto_load_depth (optional, default 2)

Sync cursor validation

Views configured with use_sync_cursor_mode: true use a cursor-based sync handler instead of receiving instances in the workflow payload. The trigger fires as a lightweight signal; the function independently fetches all changes since the last cursor from DMS and validates them.

The payload is set by the deployed workflow (populated by deploy_validation_infrastructure()) and patched with initial_sync_cutoff by the orchestrator when it creates the sync trigger. You do not normally invoke this handler directly.

Key payload parameters:

• view_space, view_external_id, view_version
• instance_space
• ruleset_references
• datamodel_space, datamodel_external_id, datamodel_version
• chunk_size – instances per sync batch
• auto_load_depth (optional, default 2)
• data_quality_space – space for cursor state storage
• initial_sync_cutoff – ISO timestamp; first-run filter to skip instances predating historic validation (patched by orchestrator)
• records_config

Aggregate uniqueness (shacl / aggregate_uniqueness)

Production uniqueness runs on the scheduled dq-{view}-uniqueness workflow with validation_type: shacl. The handler loads SHACL from RuleSet refs (or inline shacl_rules), discovers dqs:uniquenessConstraint / dqs:unique, and runs the aggregate executor.

from cognite_data_quality import call_validate_shacl

result = call_validate_shacl(
    client,
    data={
        "ruleset_references": [{"externalId": "my-ruleset", "version": "1.0.0"}],
        "instance_spaces": ["my_instances"],
        "datamodel_space": "my_space",
        "datamodel_external_id": "WorkOrder",
        "datamodel_version": "v1",
        "records_config": {
            "stream_id": "dq_validation_stream",
            "rule_set_id": "Uniqueness_WorkOrderNumber",
            "rule_set_version": "1.0",
        },
    },
)

Key payload parameters:

• ruleset_references or inline shacl_rules
• datamodel_space, datamodel_external_id, datamodel_version (for SHACL discovery)
• instance_spaces — scope aggregate and detail queries to these spaces
• records_config — posts RuleEngineResult records per violating instance

Use call_validate_aggregate_uniqueness() only when calling the low-level executor directly with an explicit property/view payload (no SHACL discovery). See Uniqueness.

Time series validation

For time series validation, the payload typically includes:

• ruleset_references or inline shacl_rules
• datamodel_space, datamodel_external_id, datamodel_version
• instance_ids (list of {"space", "external_id"}) or filter with space and/or external_id_prefix
• records_config (rule_set_id = view or RuleSet external ID; rule_set_version = RuleSet reference version from ruleset_references)

The handler returns timeseries_count and records_posted (not instances_validated).

For concrete YourOrgTimeSeries datapoint examples and an ad-hoc validation walkthrough, see Time series datapoint rule examples.

RuleEngineResult incremental consumption (rule_engine_result_sync)

Use this validation type when downstream rules or pipelines should consume only new/changed RuleEngineResult records.

Required payload:

• validation_type: "rule_engine_result_sync"
• listener_id
• stream_id

Optional payload:

• source_mode: "sync" (default) or "filter"
• filter: rule-scoped filters
• rule_ids
• rule_set_ids
• rule_set_versions
• focus_nodes
• focus_node_instances
• records_space (default dataQuality)
• records_container (default RuleEngineResult)
• initialize_cursor (default 7d-ago, sync mode only when no saved cursor exists)
• initial_watermark_ms (filter mode bootstrap; milliseconds since epoch)
• include_items (default false) — return full record payloads; probes and tests should set true
• limit (legacy alias for max_records)
• page_limit (default 1000, max 1000)
• max_records (default 10000) — enables multi-page reads (>1000) in one run

The handler persists cursor/watermark state per listener_id in FunctionValidationState and returns record_external_ids (and items when include_items is true).

Notes:

• sync mode advances by API cursor only; watermark does not drive progression.
• filter mode always uses lastUpdatedTime and advances watermark to the max seen timestamp.
• For first-run filter mode, use a bounded initial_watermark_ms (for example now-6d) when your environment enforces max interval windows.

For a full cross-run chaining walkthrough, see Chained conditional logic.

Best practices

• Prefer specific helper wrappers for readability, then fall back to call_validation() for advanced routing.
• Keep payloads sourced from deployed config artifacts where possible to reduce drift.
• Use wait=True for scripts that need immediate result inspection.

Troubleshooting

• Invocation fails: ensure infrastructure/function is deployed and credentials allow function calls.
• Handler returns empty outputs: verify payload keys for selected validation_type.
• Unexpected rule scope: check datamodel/view identifiers and rules source fields.
• View-only deploy_validation_pipeline(view_space=..., view_external_id=...) raises "No managed DataProducts configured": this mode reads the managed DataProduct list (managed_data_product_external_ids) that deploy_validation_infrastructure() writes to DMS. Run deploy_validation_infrastructure() first, or pass data_product_external_id explicitly.

Related guides

• Confirm that rules work
• Run validation locally
• Deploy infrastructure and pipeline
• Chained conditional logic

Previous section

• Records output

Next section

• Runtime options