Deploy infrastructure and pipeline

Deploy validation infrastructure (containers, Cognite Function, workflows, triggers) and run the validation pipeline for a view (historic partitions, sync trigger, monitor schedule).

For Data Product publish semantics, versioning, data_product_sync, and CI/CD patterns, see Lifecycle and CI/CD. For local validation without deployment, see Run validation locally.

Quick flow

1. Develop rules – Use run_validation() to validate against DMS. Iterate on rules until results are acceptable.
2. Deploy – Use deploy_validation_infrastructure() to deploy the function, workflows, triggers, and data quality containers.
3. Pipeline (optional) – Use deploy_validation_pipeline() to run historic validation for a view and set up sync and monitor schedules.
4. Invoke – Use the invoke helpers to call deployed Cognite Functions from Python when needed.

Rule persistence policy: see Rule sources.

Deploy validation infrastructure

Deploy all validation infrastructure for an environment in one call:

from pathlib import Path
from cognite_data_quality import deploy_validation_infrastructure, load_cognite_client_from_toml

client = load_cognite_client_from_toml("config.toml")

config_root = Path("config/environments/my_env")
settings_path = config_root / "settings.yaml"
views_dir = config_root / "views"

# Orchestrator needs credentials to create triggers
function_secrets = {
    "client-id": "your-client-id",
    "client-secret": "your-client-secret",
}

deploy_validation_infrastructure(
    client=client,
    settings_path=settings_path,
    views_dir=views_dir,
    function_secrets=function_secrets,
    force=False,
    dry_run=False,
)

This call:

• Ensures Records API container (dataQuality/DataQualityValidationRecord)
• Ensures the rule-engine container (dataQuality/RuleEngineResult)
• Ensures record views DataQualityValidationRecord and RuleEngineResult version v1 (with streamId) in one apply — not attached to a data model
• Ensures state containers (OrchestrationState, FunctionValidationState) and deploys the monitoring data model and views
• Deploys the unified validation function (with embedded handler code)
• Publishes local SHACL TTL to the RuleSet API and view configs to the DataProduct API (when dataproducts: is set on views/timeseries YAML)
• Persists deployment settings to DMS (DataQualitySettings, including managed_data_product_external_ids and timeseries_workflow_external_id_prefix)
• Deploys consolidated instance/uniqueness workflows per DataProduct via data_product_sync or external_dataproducts
• Deploys time series validation workflows when timeseries_dir is provided
• Uploads SHACL rules to CDF Files for RAW tables only (tables/ configs)
• Deploys rule_engine_result_sync listener workflows when configured
• Optionally deploys the packaged Atlas rule-authoring skill + agent when atlas_rule_authoring.enabled: true or deploy_atlas_rule_authoring=True (see Atlas rule-authoring agent)

Instance and time series SHACL are not uploaded to CDF Files in 0.4.0 — workflow payloads use ruleset_references only.

Time series datapoint tests should be scheduled in production. Use the environment timeseries config cron cadence and validate rule behavior with the scripts documented in Time series datapoint rule examples.

You can extract credentials from the client when available:

creds = client.config.credentials
if hasattr(creds, "client_id") and hasattr(creds, "client_secret"):
    function_secrets = {"client-id": creds.client_id, "client-secret": creds.client_secret}
else:
    function_secrets = None  # Orchestrator triggers will fail without secrets

Deploy validation pipeline

After infrastructure is deployed, run the full validation pipeline for a view (historic partitions, sync trigger, monitor schedule). See Sync-cursor and historic orchestration for the production lifecycle.

From CDF Files (orchestrator invoke shorthand)

deploy_validation_pipeline() can resolve view and settings YAML from CDF Files when you pass view_config_external_id (and optionally settings_external_id). This is a convenience for invoking the historic orchestrator without local config files — it does not load instance SHACL from CDF Files. The orchestrator payload must include ruleset_references (populated automatically when using local YAML with DataProduct deploy, or from view YAML that lists RuleSet refs).

from cognite_data_quality import deploy_validation_pipeline

result = deploy_validation_pipeline(
    client,
    view_config_external_id="Pump_view_config",  # view YAML in CDF Files (if uploaded)
)

Settings are fetched from data-quality-validation_settings in CDF Files when settings_path is not provided. Override with settings_external_id if needed:

result = deploy_validation_pipeline(
    client,
    view_config_external_id="Pump_view_config",
    settings_external_id="my-function_settings",
)

Prefer local YAML + settings_path for new deployments. See Migration to 0.4.0.

From local YAML files

result = deploy_validation_pipeline(
    client,
    settings_path=str(settings_path),
    view_space="my_space",       # when external_id is ambiguous across spaces
    view_external_id="MyView",
    wait=True,
)
# result: orchestration_id, status, partitions_triggered, partition_count,
#         sync_trigger_external_id, monitor_schedule_name, distribution, ...

Settings and view config are read from settings_path and the corresponding view YAML under the same environment. The pipeline triggers partitioned validation for historic data and sets up the sync trigger and monitor schedule for incremental runs.

DataProduct mode (view-only resolve)

After deploy_validation_infrastructure(), managed DataProduct external IDs are written to DMS. Historic orchestration can be triggered with only the view identity — no data_product_external_id:

from cognite_data_quality import deploy_validation_pipeline

deploy_validation_pipeline(
    client,
    view_space="my_space",
    view_external_id="Equipment",
)

The function scans managed_data_product_external_ids in DMS, loads each DataProduct version from the API, and matches on both view.space and view.external_id. Pass view_space when the same external ID exists in multiple schema spaces. Pass data_product_external_id when the view is listed in more than one managed DataProduct.

For externally-owned DataProducts, local view YAML overrides (e.g. instance_spaces) are matched by (view.space, view.external_id) — not by dataproducts: on the YAML. Declare external DPs in settings.external_dataproducts.

Historic queue enqueue mode (0.4.4+)

For when to use historic enqueue versus local run_validation(), including force=True without a DataProduct version bump, see Confirm that rules work. Historic always uses published RuleSets; unpublished TTL is not applied.

Use historic_mode="enqueue" when you want sequential historic processing via HistoricJobQueue and dq-historic-queue-manager instead of parallel partition orchestrations per view.

from cognite_data_quality import deploy_validation_pipeline, enqueue_historic_validation

# Enqueue all managed DataProducts (from DMS managed_data_product_external_ids)
enqueue_historic_validation(
    client,
    data_quality_space="sp_data_quality",
    trigger_queue_manager=True,  # optional one-shot queue manager run
)

# Same via deploy_validation_pipeline (DataProduct mode)
deploy_validation_pipeline(
    client,
    data_product_external_id="my-data-product",
    historic_mode="enqueue",
    trigger_queue_manager=True,
)

# Re-run historic for all ruled views without a DataProduct/RuleSet version bump,
# including views with historic_reprocess_on_update: false.
# trigger_queue_manager=True starts dq-historic-queue-manager; without it the
# job stays queued until the next cron.
enqueue_historic_validation(
    client,
    data_product_external_id="my-data-product",
    force=True,
    trigger_queue_manager=True,
)

When to use enqueue vs orchestrator:

Mode	Processing	Sync trigger	Typical use
orchestrator (default)	Parallel partition workers per view	Created by orchestrator Step 3	Manual one-shot historic + incremental handoff
enqueue	Sequential via HistoricJobQueue	Created when queue manager runs orchestrator for each job	CI after deploy, estates with many views, idempotent re-runs

Enqueue is idempotent per (view, data_product_version) — completed jobs for the same version are skipped; stale active jobs are superseded before a new enqueue. Pass force=True to re-queue a completed historic run without a DataProduct or RuleSet version bump, including views with historic_reprocess_on_update: false. Jobs that are already queued or running are still skipped. Use view_external_id only if you want to limit the force-run to specific views.

force=True on deploy_validation_pipeline requires historic_mode="enqueue" — the default orchestrator mode raises rather than ignoring force.

Do not confuse the two force flags:

Flag	Effect
data_product_sync payload force: true	Redeploy workflows for the current DataProduct version. Historic enqueue stays idempotent (completed same-version jobs are skipped). Does not bypass historic_reprocess_on_update: false.
enqueue_historic_validation(..., force=True)	Re-queue completed historic jobs for the current version, including opted-out views. Does not redeploy workflows.

data_product_sync also enqueues historic jobs when it deploys workflows for a new DataProduct version. If workflows already carry the correct [dp:{version}] tag, sync skips deploy and does not enqueue — use enqueue_historic_validation() after infrastructure deploy in that case. Use enqueue force=True when you need a new historic scan of the current version.

Return shape differs by mode — check mode on the result:

• historic_mode="enqueue" → dict with mode: "enqueue", status, data_products, total_enqueued, total_skipped, optional warnings, optional queue_manager_execution_id
• historic_mode="orchestrator" (default) → list of per-view dicts with view, view_space, data_product, call_id, status

See Migration 0.4.4 and Sync-cursor and historic orchestration.

Advanced parameters

Runtime credentials and fallback

You can keep using function_secrets, or define grouped runtime credentials in settings.yaml:

runtime_credentials:
  functions:
    client_id: ${DQ_FUNCTION_CLIENT_ID}
    client_secret: ${DQ_FUNCTION_CLIENT_SECRET}
  workflows:
    client_id: ${DQ_WORKFLOW_CLIENT_ID}
    client_secret: ${DQ_WORKFLOW_CLIENT_SECRET}
  schedules:
    client_id: ${DQ_SCHEDULE_CLIENT_ID}
    client_secret: ${DQ_SCHEDULE_CLIENT_SECRET}

Resolution order:

1. Explicit grouped credentials for the target (functions, workflows, schedules)
2. Deployment client credentials (client.config.credentials)
3. For function deployment secrets only: legacy env fallback COGNITE_CLIENT_ID / COGNITE_CLIENT_SECRET (or IDP_*)

This keeps old deployments backward-compatible while allowing credential separation.

Records stream type and permissions

Validation records are written to a CDF Records API stream whose mutability is fixed when the stream is created and cannot be changed. Set records.stream_type in settings.yaml to the CDF template name used when creating a new stream:

records:
  stream_id: dq_validation_stream
  stream_type: BasicArchive   # BasicArchive (default) | BasicLiveData | ImmutableTestStream

• BasicArchive: immutable, append/history. Every run is kept; queries require a lastUpdatedTime range.
• BasicLiveData: mutable, current-state. Records upsert to one row per stable key; de-duplicated statistics.
• ImmutableTestStream: immutable test template.

records.stream_type only affects stream creation (persisted in DMS as records_stream_type). At runtime the validation function auto-detects the actual stream mutability (via GET /streams/{id}) and dispatches to POST /records (ingest) or POST /records/upsert accordingly.

deploy_validation_infrastructure() ensures the Records container and record views (DataQualityValidationRecord and RuleEngineResult version v1, with streamId). It does not create the stream. Create the stream separately (CDF Console/API) with the chosen template, or use ensure_records_infrastructure() / local run_validation(post_to_records=True) which creates the stream if missing. If the stream is still missing, view apply is skipped with a warning and the rest of infrastructure deploy continues. See Records streams → Creating the stream and Records output → Record views.

Stream type auto-detection. Any GET /streams/{id} failure fail-safes to immutable ingest (not only missing streamsAcl:READ). Grant streamsAcl:READ before using a mutable stream. See Records output → Stream mutability.

Consumer coordination. Records use a stable externalId (no jobRunId in the key). On immutable streams the ID is no longer unique per run; consumers must filter by jobRunId / lastUpdatedTime. The Streamlit dashboard still assumes immutable/run-scoped semantics — do not switch a production stream to BasicLiveData until the consumer path is updated. See Records output → Stream mutability.

Record views (0.5)

On 0.5.x, deploy ensures the two record views in one views.apply([...]) (identifiers and Flows contract: Records output → Record views (Flows)). Opt-in instance → records YAML lives only in Reverse navigation from instance views (0.5.x contract and warnings are documented there).

Function secrets (explicit override)

function_secrets still takes precedence when passed directly:

deploy_validation_infrastructure(
    client=client,
    settings_path=settings_path,
    views_dir=views_dir,
    function_secrets={"client-id": "...", "client-secret": "..."},
)

The orchestrator now supports optional specialized runtime secret keys in addition to legacy keys:

• Legacy fallback keys: client-id, client-secret
• Workflow trigger keys: workflow-client-id, workflow-client-secret
• Function schedule keys: schedule-client-id, schedule-client-secret

Custom function external ID

Override the default function external ID (e.g. when multiple environments share a project):

deploy_validation_infrastructure(
    client=client,
    settings_path=settings_path,
    views_dir=views_dir,
    function_external_id="data-quality-validation-myenv",
)

Time series and SHACL dirs

Optional directories for time series configs and SHACL rules:

deploy_validation_infrastructure(
    client=client,
    settings_path=settings_path,
    views_dir=views_dir,
    timeseries_dir=config_root / "timeseries",
    shacl_rules_dir=config_root / "shacl_rules",
)

Dry run and force

• dry_run=True – Log what would be deployed without making changes.
• force=True – Redeploy even when content hashes match (useful after dependency or config changes).

For Data Product publishing, data_product_sync, listener workflows, and externally-owned DataProducts, see Lifecycle and CI/CD.

Incremental deployment (single view)

For deploying only the function and workflow for a single view config, see the API reference for deploy_incremental(). It takes a view config path and optional config_env; the main recommended path is deploy_validation_infrastructure() plus deploy_validation_pipeline() as above.

Scheduled uniqueness workflows

Global uniqueness is declared in SHACL/RuleSet — not in view YAML. When deploy scans TTL (local file or fetched RuleSet chunks) and finds dqs:uniquenessConstraint or dqs:unique for a view's target class, it creates:

Resource	Pattern
Workflow	dq-{view}-uniqueness
Trigger	cron from dqs:schedule, optional YAML uniqueness_cron override, or default 0 6 * * *
Function payload	validation_type: shacl with same rules ref as the sync workflow

Views without uniqueness constraints in SHACL get sync/incremental workflows only.

Sync-cursor and instance validation handlers do not run uniqueness after each sync — only the scheduled workflow (or an explicit call_validate_shacl() / validation_type: shacl invoke).

Current record behavior for uniqueness failures:

• Runtime suppresses repeated failure records for unchanged instances.
• If a prior failed record for the same focusNode is newer/equal to the instance lastUpdatedTime, no new failure record is written.
• Once the instance changes, failure records are written again if the violation remains.

Optional YAML override to disable or override schedule:

# views/my_view.yaml — deployment mechanics only; rules stay in TTL/RuleSet
uniqueness_cron: null          # disable scheduled uniqueness workflow
# uniqueness_cron: "0 * * * *" # override when SHACL has no dqs:schedule

See Uniqueness and Rule sources for SHACL authoring.

Edge existence workflows (typed edge cardinality)

Typed-edge cardinality is declared in SHACL/RuleSet with dqs:edgeExistenceConstraint. When deploy / data_product_sync finds constraints for a DataProduct view, it creates:

Resource	Pattern
Workflow	{instance_workflow_id}-edge-existence
Task	validation_type: edge_existence_event
Triggers	2 per rule: DMS node sync + DMS edge sync (no_backfill)

Instance sync triggers remain orchestrator-only (sync-cursor guide). Edge existence triggers are deploy-time. On DataProduct/RuleSet version-change historic runs, orchestrator Step 3b runs edge_existence_baseline over all in-scope focus nodes.

See Edge existence and Rule sources for SHACL authoring.

Atlas rule-authoring skill and agent

Opt-in. Definitions ship in the cognite-data-quality wheel. Enable with atlas_rule_authoring.enabled: true in settings.yaml, or pass deploy_atlas_rule_authoring=True to deploy_validation_infrastructure().

atlas_rule_authoring:
  enabled: true
  agent_external_id: DQ_AGENT_0001
  verify: true

Standalone deploy (without full infrastructure):

from cognite_data_quality import deploy_atlas_rule_authoring

deploy_atlas_rule_authoring(client)

Full procedure, chat authoring notes, and boundaries: Atlas rule-authoring agent.

Best practices

• Keep deployment config and local validation config aligned to avoid behavior drift.
• Treat DataProduct/RuleSet publishing as idempotent and payload-aware — see Lifecycle and CI/CD.
• Use scheduled uniqueness and listener workflows for global checks and chaining.
• Plan CDF trigger budget when enabling many dqs:edgeExistenceConstraint rules (two triggers each).

Troubleshooting

• Deploy succeeds but workflows do not run: verify secrets/credentials and trigger schedules.
• Duplicate lifecycle behavior unexpected: verify version existence, payload reuse, and requested semver.
• Missing uniqueness workflow: ensure SHACL declares uniqueness for target class/property.
• Missing edge-existence workflow: ensure RuleSet SHACL has dqs:edgeExistenceConstraint with {space}/{externalId} dqs:edgeType.

Related guides

• Run validation locally
• Lifecycle and CI/CD
• Invoke Cognite Functions
• Rule sources
• Atlas rule-authoring agent
• Reverse navigation from instance views
• Records output

Previous section

• Runtime options

Next section

• Lifecycle and CI/CD