Invoke

The cognite_data_quality.invoke module provides helpers for calling already-deployed Cognite Functions from Python. These helpers send a payload to the function running in CDF — they do not run validation locally.

• call_validation() — unified entry point, dispatches by validation_type
• call_validate_instances_shacl() — instance validation (type: instance)
• call_validate_instances_shacl_partitioned() — partitioned worker (type: partitioned)
• call_validate_timeseries_shacl() — time series validation (type: timeseries)
• call_validate_aggregate_uniqueness() — aggregate uniqueness (type: aggregate_uniqueness)
• call_validate_shacl() — SHACL-centric entrypoint with extension support (type: shacl)
• enqueue_historic_validation() — enqueue historic jobs on HistoricJobQueue (type: queue enqueue; sequential processing)
• deploy_validation_pipeline() — deploy and run the full validation pipeline (historic_mode: orchestrator or enqueue)

See Invoking validation functions for usage examples and payload key descriptions. For local sample vs estate historic (force=True), see Confirm that rules work.

cognite_data_quality.invoke

Invoke deployed Cognite Functions from Python.

Use these helpers to call the data quality validation functions from a notebook, script, or any Python runtime. Each function is a thin wrapper around client.functions.call().

call_rule_engine_result_sync(client, data, *, wait=True, external_id='data-quality-validation')

Call the incremental RuleEngineResult sync handler.

Source code in cognite_data_quality/invoke.py

def call_rule_engine_result_sync(
    client: CogniteClient,
    data: dict,
    *,
    wait: bool = True,
    external_id: str = "data-quality-validation",
) -> dict:
    """Call the incremental RuleEngineResult sync handler."""
    return call_validation(client, "rule_engine_result_sync", data, wait=wait, external_id=external_id)

call_validate_aggregate_uniqueness(client, data, *, wait=True, external_id='data-quality-validation')

Call the aggregate uniqueness validation handler.

Convenience wrapper for call_validation(validation_type="aggregate_uniqueness", ...).

Source code in cognite_data_quality/invoke.py

def call_validate_aggregate_uniqueness(
    client: CogniteClient,
    data: dict,
    *,
    wait: bool = True,
    external_id: str = "data-quality-validation",
) -> dict:
    """Call the aggregate uniqueness validation handler.

    Convenience wrapper for call_validation(validation_type="aggregate_uniqueness", ...).
    """
    return call_validation(client, "aggregate_uniqueness", data, wait=wait, external_id=external_id)

call_validate_instances_shacl(client, data, *, wait=True, external_id='data-quality-validation')

Call the instance SHACL validation function.

Convenience wrapper for call_validation(validation_type="instance", ...).

Parameters:

Name	Type	Description	Default
client	CogniteClient	Cognite client instance.	required
data	dict	Payload dict. Typical keys: instances (dict with items), shacl_rules or ruleset_references, datamodel_space, datamodel_external_id, datamodel_version, auto_load_depth, records_config (stream_id, rule_set_id, etc.).	required
wait	bool	If True, block until the function returns.	True
external_id	str	Function external ID (default: data-quality-validation).	'data-quality-validation'

Returns:

Type	Description
dict	Response dict with conforms, violations, instances_validated, etc.

Source code in cognite_data_quality/invoke.py

def call_validate_instances_shacl(
    client: CogniteClient,
    data: dict,
    *,
    wait: bool = True,
    external_id: str = "data-quality-validation",
) -> dict:
    """Call the instance SHACL validation function.

    Convenience wrapper for call_validation(validation_type="instance", ...).

    Args:
        client: Cognite client instance.
        data: Payload dict. Typical keys: instances (dict with items), shacl_rules or
            ruleset_references, datamodel_space, datamodel_external_id,
            datamodel_version, auto_load_depth, records_config (stream_id, rule_set_id, etc.).
        wait: If True, block until the function returns.
        external_id: Function external ID (default: data-quality-validation).

    Returns:
        Response dict with conforms, violations, instances_validated, etc.
    """
    return call_validation(client, "instance", data, wait=wait, external_id=external_id)

call_validate_instances_shacl_partitioned(client, data, *, wait=True, external_id='data-quality-validation')

Call the partitioned instance SHACL validation function (worker).

Convenience wrapper for call_validation(validation_type="partitioned", ...).

Parameters:

Name	Type	Description	Default
client	CogniteClient	Cognite client instance.	required
data	dict	Payload dict. Same shape as validate-instances-shacl; instances are typically provided by the orchestrator.	required
wait	bool	If True, block until the function returns.	True
external_id	str	Function external ID (default: data-quality-validation).	'data-quality-validation'

Returns:

Type	Description
dict	Response dict with status, conforms, violations, instances_validated, etc.

Source code in cognite_data_quality/invoke.py

def call_validate_instances_shacl_partitioned(
    client: CogniteClient,
    data: dict,
    *,
    wait: bool = True,
    external_id: str = "data-quality-validation",
) -> dict:
    """Call the partitioned instance SHACL validation function (worker).

    Convenience wrapper for call_validation(validation_type="partitioned", ...).

    Args:
        client: Cognite client instance.
        data: Payload dict. Same shape as validate-instances-shacl; instances are
            typically provided by the orchestrator.
        wait: If True, block until the function returns.
        external_id: Function external ID (default: data-quality-validation).

    Returns:
        Response dict with status, conforms, violations, instances_validated, etc.
    """
    return call_validation(client, "partitioned", data, wait=wait, external_id=external_id)

call_validate_pre_persist(client, data, *, wait=True, external_id='data-quality-validation')

Validate a proposed DMS patch before persisting it.

Source code in cognite_data_quality/invoke.py

def call_validate_pre_persist(
    client: CogniteClient,
    data: dict,
    *,
    wait: bool = True,
    external_id: str = "data-quality-validation",
) -> dict:
    """Validate a proposed DMS patch before persisting it."""
    return call_validation(client, "pre_persist", data, wait=wait, external_id=external_id)

call_validate_shacl(client, data, *, wait=True, external_id='data-quality-validation')

Call the SHACL-centric validation entrypoint with optional global-rule extensions.

Convenience wrapper for call_validation(validation_type="shacl", ...).

Source code in cognite_data_quality/invoke.py

def call_validate_shacl(
    client: CogniteClient,
    data: dict,
    *,
    wait: bool = True,
    external_id: str = "data-quality-validation",
) -> dict:
    """Call the SHACL-centric validation entrypoint with optional global-rule extensions.

    Convenience wrapper for call_validation(validation_type="shacl", ...).
    """
    return call_validation(client, "shacl", data, wait=wait, external_id=external_id)

call_validate_timeseries_shacl(client, data, *, wait=True, external_id='data-quality-validation')

Call the time series SHACL validation function.

Convenience wrapper for call_validation(validation_type="timeseries", ...).

Parameters:

Name	Type	Description	Default
client	CogniteClient	Cognite client instance.	required
data	dict	Payload dict. Typical keys: ruleset_references or shacl_rules, datamodel_space, datamodel_external_id, datamodel_version, filter or instance_ids, records_config.	required
wait	bool	If True, block until the function returns.	True
external_id	str	Function external ID (default: data-quality-validation).	'data-quality-validation'

Returns:

Type	Description
dict	Response dict with validation results.

Source code in cognite_data_quality/invoke.py

def call_validate_timeseries_shacl(
    client: CogniteClient,
    data: dict,
    *,
    wait: bool = True,
    external_id: str = "data-quality-validation",
) -> dict:
    """Call the time series SHACL validation function.

    Convenience wrapper for call_validation(validation_type="timeseries", ...).

    Args:
        client: Cognite client instance.
        data: Payload dict. Typical keys: ruleset_references or shacl_rules, datamodel_space,
            datamodel_external_id, datamodel_version, filter or instance_ids,
            records_config.
        wait: If True, block until the function returns.
        external_id: Function external ID (default: data-quality-validation).

    Returns:
        Response dict with validation results.
    """
    return call_validation(client, "timeseries", data, wait=wait, external_id=external_id)

call_validation(client, validation_type, data, *, wait=True, external_id='data-quality-validation')

Call the unified SHACL validation function.

The unified function consolidates all validation types into a single deployment. Use this function for new code. Legacy individual functions are still supported.

Parameters:

Name	Type	Description	Default
client	CogniteClient	Cognite client instance.	required
validation_type	Literal['instance', 'partitioned', 'timeseries', 'orchestrator', 'test', 'aggregate_uniqueness', 'rule_engine_result_sync', 'shacl', 'pre_persist']	Type of validation to perform: - "instance": Basic instance validation (real-time) - "partitioned": Partitioned historic instance validation - "timeseries": Time series validation (normal/backfill) - "orchestrator": Historic validation orchestrator - "test": Test SHACL rules without writing to Records API - "aggregate_uniqueness": Aggregate-based uniqueness validation - "rule_engine_result_sync": Incremental RuleEngineResult listener - "shacl": SHACL-centric entrypoint with optional global-rule extensions - "pre_persist": Validate a proposed DMS patch before apply	required
data	dict	Payload dict. Contents depend on validation_type.	required
wait	bool	If True, block until the function returns.	True
external_id	str	Function external ID (default: data-quality-validation).	'data-quality-validation'

Returns:

Type	Description
dict	Response dict from the appropriate handler.

Example

Instance validation

result = call_validation( client=client, validation_type="instance", data={ "instances": {"items": [...]}, "ruleset_references": [{"externalId": "my_rules", "version": "1.0.0"}], "datamodel_space": "my_space", "datamodel_external_id": "MyModel", "datamodel_version": "v1", "records_config": {...} } )

Orchestrator

result = call_validation( client=client, validation_type="orchestrator", data={ "datamodel_space": "my_space", "view_external_id": "MyView", "instance_space": "my_instances", ... } )

Source code in cognite_data_quality/invoke.py

def call_validation(
    client: CogniteClient,
    validation_type: Literal[
        "instance",
        "partitioned",
        "timeseries",
        "orchestrator",
        "test",
        "aggregate_uniqueness",
        "rule_engine_result_sync",
        "shacl",
        "pre_persist",
    ],
    data: dict,
    *,
    wait: bool = True,
    external_id: str = "data-quality-validation",
) -> dict:
    """Call the unified SHACL validation function.

    The unified function consolidates all validation types into a single deployment.
    Use this function for new code. Legacy individual functions are still supported.

    Args:
        client: Cognite client instance.
        validation_type: Type of validation to perform:
            - "instance": Basic instance validation (real-time)
            - "partitioned": Partitioned historic instance validation
            - "timeseries": Time series validation (normal/backfill)
            - "orchestrator": Historic validation orchestrator
            - "test": Test SHACL rules without writing to Records API
            - "aggregate_uniqueness": Aggregate-based uniqueness validation
            - "rule_engine_result_sync": Incremental RuleEngineResult listener
            - "shacl": SHACL-centric entrypoint with optional global-rule extensions
            - "pre_persist": Validate a proposed DMS patch before apply
        data: Payload dict. Contents depend on validation_type.
        wait: If True, block until the function returns.
        external_id: Function external ID (default: data-quality-validation).

    Returns:
        Response dict from the appropriate handler.

    Example:
        # Instance validation
        result = call_validation(
            client=client,
            validation_type="instance",
            data={
                "instances": {"items": [...]},
                "ruleset_references": [{"externalId": "my_rules", "version": "1.0.0"}],
                "datamodel_space": "my_space",
                "datamodel_external_id": "MyModel",
                "datamodel_version": "v1",
                "records_config": {...}
            }
        )

        # Orchestrator
        result = call_validation(
            client=client,
            validation_type="orchestrator",
            data={
                "datamodel_space": "my_space",
                "view_external_id": "MyView",
                "instance_space": "my_instances",
                ...
            }
        )
    """
    payload = {"validation_type": validation_type, **data}
    call = client.functions.call(external_id=external_id, data=payload, wait=wait)
    return call.get_response() or {}

deploy_validation_pipeline(client, data=None, *, settings_path=None, views_dir=None, view_external_id=None, view_space=None, view_config_external_id=None, settings_external_id='data-quality-validation_settings', data_product_external_id=None, data_product_version=None, data_quality_space='dataQuality', wait=True, external_id='data-quality-validation', historic_mode='orchestrator', trigger_queue_manager=False, queue_manager_workflow_external_id='dq-historic-queue-manager', queue_manager_workflow_version='1', force=False)

Deploy and run a full validation pipeline with batch and incremental validation.

This function: 1. Triggers partitioned batch load of all instances (splits data into partitions) 2. Sets up sync trigger for ongoing incremental validation 3. Creates scheduled workflow for monitoring progress 4. Continuously manages orchestration (retries, cleanup)

Convenience wrapper for call_validation(validation_type="orchestrator", ...).

Three usage modes:

Mode 1: Pass data dict directly (legacy): deploy_validation_pipeline(client, data={...}, wait=True)

Mode 2: Load from local YAML configs: deploy_validation_pipeline( client, settings_path="config/environments/example/settings.yaml", view_external_id="SmallBoat", wait=True )

Mode 3: Load everything from CDF Files (no local files required): deploy_validation_pipeline( client, view_config_external_id="pump_view_config", )

Mode 4: DataProduct mode — triggers historic orchestrator per view: deploy_validation_pipeline( client, data_product_external_id="dp_rmdm_dm", view_external_id="Pump", # omit to run all views data_quality_space="sp_data_quality_demo", )

# View-only resolve (DataProduct inferred from managed_data_product_external_ids in DMS):
deploy_validation_pipeline(
    client,
    view_space="my_space",
    view_external_id="Pump",
)
Loads view configs from the DataProduct API and calls the orchestrator for
each matching view. The orchestrator runs the historic batch validation and
sets up the per-view sync trigger on completion.
Infrastructure (``data_product_sync`` schedule, workflow definitions) must be
deployed first via ``deploy_validation_infrastructure()``.

Mode 4 enqueue — queue historic jobs instead of parallel orchestrators::

deploy_validation_pipeline(
    client,
    data_product_external_id="dp_rmdm_dm",
    data_quality_space="sp_data_quality_demo",
    historic_mode="enqueue",
    trigger_queue_manager=True,
)

Uses ``enqueue_historic_validation()``; ``historic_queue_manager`` processes
the queue sequentially. Completed jobs for the same DP version are skipped
unless ``force=True``.

Parameters:

Name	Type	Description	Default
client	CogniteClient	Cognite client instance.	required
data	dict | None	(Legacy) Payload dict. Typical keys: datamodel_space, datamodel_external_id, datamodel_version, view_space, view_external_id, view_version, instance_space, ruleset_references, partition_count, partition_field, records_space, records_container, stream_id, rule_set_id, rule_set_version. If None, must provide settings_path or view_config_external_id.	None
settings_path	str | None	Path to a local settings.yaml file. Optional when using view_config_external_id — settings are then fetched from CDF Files instead.	None
views_dir	str | None	Directory containing view YAML configs. Defaults to settings_path parent / "views". Only used in mode 2.	None
view_external_id	str | list[str] | None	Specific view external ID (modes 2 and 4). Use with view_space when the same external ID exists in multiple schema spaces.	None
view_space	str | None	DMS schema space for the view (modes 2 and 4). Required when the external ID is ambiguous across spaces.	None
view_config_external_id	str | None	External ID of a view config stored in CDF Files (mode 3). Uploaded during deployment via deploy_instance_workflows(). External ID scheme: {view_external_id}_view_config e.g. Pump_view_config.	None
settings_external_id	str	External ID of the settings file stored in CDF Files. Defaults to data-quality-validation_settings (the ID used by deploy_instance_workflows). Only used in mode 3 when settings_path is not provided.	'data-quality-validation_settings'
data_quality_space	str	DMS space where the DataQualitySettings node lives. Must match config_space (or records.space when config_space is unset) in settings.yaml. Default: "dataQuality". Only used in mode 4.	'dataQuality'
wait	bool	If True, block until the function returns.	True
external_id	str	Function external ID (default: data-quality-validation). Only used in legacy mode (when data dict is provided directly).	'data-quality-validation'
historic_mode	Literal['orchestrator', 'enqueue']	In DataProduct mode (mode 4), "orchestrator" triggers the orchestrator immediately per view (parallel). "enqueue" writes HistoricJobQueue jobs for sequential processing.	'orchestrator'
trigger_queue_manager	bool	When historic_mode="enqueue", optionally start dq-historic-queue-manager once after enqueue.	False
queue_manager_workflow_external_id	str	Workflow external ID for the queue manager.	'dq-historic-queue-manager'
queue_manager_workflow_version	str	Workflow version for the queue manager.	'1'
force	bool	When historic_mode="enqueue", re-queue completed jobs for the same DataProduct/RuleSet version, including views with historic_reprocess_on_update=false. Raises ValueError if force=True and historic_mode is not "enqueue". Does not redeploy workflows.	False

Returns:

Type	Description
dict	Response dict from the function.

Example

Mode 4: DataProduct mode — trigger historic validation for one view

result = deploy_validation_pipeline( client, data_product_external_id="dp_rmdm_dm", view_external_id="Pump", # omit for all views data_quality_space="sp_data_quality_demo", )

Mode 3: fully from CDF — no local files needed

result = deploy_validation_pipeline( client, view_config_external_id="pump_view_config", )

Mode 2: local YAML files

result = deploy_validation_pipeline( client, settings_path="config/environments/example/settings.yaml", view_external_id="SmallBoat", )

Mode 1: legacy data dict

result = deploy_validation_pipeline( client, data={"datamodel_space": "sailboat", ...} )

Source code in cognite_data_quality/invoke.py

def deploy_validation_pipeline(
    client: CogniteClient,
    data: dict | None = None,
    *,
    settings_path: str | None = None,
    views_dir: str | None = None,
    view_external_id: str | list[str] | None = None,
    view_space: str | None = None,
    view_config_external_id: str | None = None,
    settings_external_id: str = "data-quality-validation_settings",
    data_product_external_id: str | None = None,
    data_product_version: str | None = None,
    data_quality_space: str = "dataQuality",
    wait: bool = True,
    external_id: str = "data-quality-validation",
    historic_mode: Literal["orchestrator", "enqueue"] = "orchestrator",
    trigger_queue_manager: bool = False,
    queue_manager_workflow_external_id: str = "dq-historic-queue-manager",
    queue_manager_workflow_version: str = "1",
    force: bool = False,
) -> dict:
    """Deploy and run a full validation pipeline with batch and incremental validation.

    This function:
    1. Triggers partitioned batch load of all instances (splits data into partitions)
    2. Sets up sync trigger for ongoing incremental validation
    3. Creates scheduled workflow for monitoring progress
    4. Continuously manages orchestration (retries, cleanup)

    Convenience wrapper for call_validation(validation_type="orchestrator", ...).

    Three usage modes:

    **Mode 1: Pass data dict directly (legacy):**
        deploy_validation_pipeline(client, data={...}, wait=True)

    **Mode 2: Load from local YAML configs:**
        deploy_validation_pipeline(
            client,
            settings_path="config/environments/example/settings.yaml",
            view_external_id="SmallBoat",
            wait=True
        )

    **Mode 3: Load everything from CDF Files (no local files required):**
        deploy_validation_pipeline(
            client,
            view_config_external_id="pump_view_config",
        )

    **Mode 4: DataProduct mode — triggers historic orchestrator per view:**
        deploy_validation_pipeline(
            client,
            data_product_external_id="dp_rmdm_dm",
            view_external_id="Pump",          # omit to run all views
            data_quality_space="sp_data_quality_demo",
        )

        # View-only resolve (DataProduct inferred from managed_data_product_external_ids in DMS):
        deploy_validation_pipeline(
            client,
            view_space="my_space",
            view_external_id="Pump",
        )
        Loads view configs from the DataProduct API and calls the orchestrator for
        each matching view. The orchestrator runs the historic batch validation and
        sets up the per-view sync trigger on completion.
        Infrastructure (``data_product_sync`` schedule, workflow definitions) must be
        deployed first via ``deploy_validation_infrastructure()``.

    **Mode 4 enqueue** — queue historic jobs instead of parallel orchestrators::

        deploy_validation_pipeline(
            client,
            data_product_external_id="dp_rmdm_dm",
            data_quality_space="sp_data_quality_demo",
            historic_mode="enqueue",
            trigger_queue_manager=True,
        )

        Uses ``enqueue_historic_validation()``; ``historic_queue_manager`` processes
        the queue sequentially. Completed jobs for the same DP version are skipped
        unless ``force=True``.

    Args:
        client: Cognite client instance.
        data: (Legacy) Payload dict. Typical keys: datamodel_space, datamodel_external_id,
            datamodel_version, view_space, view_external_id, view_version,
            instance_space, ruleset_references, partition_count, partition_field,
            records_space, records_container, stream_id, rule_set_id, rule_set_version.
            If None, must provide settings_path or view_config_external_id.
        settings_path: Path to a local settings.yaml file. Optional when using
            view_config_external_id — settings are then fetched from CDF Files instead.
        views_dir: Directory containing view YAML configs. Defaults to settings_path parent / "views".
            Only used in mode 2.
        view_external_id: Specific view external ID (modes 2 and 4). Use with
            ``view_space`` when the same external ID exists in multiple schema spaces.
        view_space: DMS schema space for the view (modes 2 and 4). Required when the
            external ID is ambiguous across spaces.
        view_config_external_id: External ID of a view config stored in CDF Files (mode 3).
            Uploaded during deployment via deploy_instance_workflows().
            External ID scheme: ``{view_external_id}_view_config`` e.g. ``Pump_view_config``.
        settings_external_id: External ID of the settings file stored in CDF Files.
            Defaults to ``data-quality-validation_settings`` (the ID used by deploy_instance_workflows).
            Only used in mode 3 when settings_path is not provided.
        data_quality_space: DMS space where the ``DataQualitySettings`` node lives.
            Must match ``config_space`` (or ``records.space`` when ``config_space`` is unset)
            in ``settings.yaml``. Default: ``"dataQuality"``. Only used in mode 4.
        wait: If True, block until the function returns.
        external_id: Function external ID (default: data-quality-validation).
            Only used in legacy mode (when ``data`` dict is provided directly).
        historic_mode: In DataProduct mode (mode 4), ``"orchestrator"`` triggers the
            orchestrator immediately per view (parallel). ``"enqueue"`` writes
            ``HistoricJobQueue`` jobs for sequential processing.
        trigger_queue_manager: When ``historic_mode="enqueue"``, optionally start
            ``dq-historic-queue-manager`` once after enqueue.
        queue_manager_workflow_external_id: Workflow external ID for the queue manager.
        queue_manager_workflow_version: Workflow version for the queue manager.
        force: When ``historic_mode="enqueue"``, re-queue completed jobs for the
            same DataProduct/RuleSet version, including views with
            ``historic_reprocess_on_update=false``. Raises ``ValueError`` if
            ``force=True`` and ``historic_mode`` is not ``"enqueue"``. Does not
            redeploy workflows.

    Returns:
        Response dict from the function.

    Example:
        # Mode 4: DataProduct mode — trigger historic validation for one view
        result = deploy_validation_pipeline(
            client,
            data_product_external_id="dp_rmdm_dm",
            view_external_id="Pump",           # omit for all views
            data_quality_space="sp_data_quality_demo",
        )

        # Mode 3: fully from CDF — no local files needed
        result = deploy_validation_pipeline(
            client,
            view_config_external_id="pump_view_config",
        )

        # Mode 2: local YAML files
        result = deploy_validation_pipeline(
            client,
            settings_path="config/environments/example/settings.yaml",
            view_external_id="SmallBoat",
        )

        # Mode 1: legacy data dict
        result = deploy_validation_pipeline(
            client,
            data={"datamodel_space": "sailboat", ...}
        )
    """
    if force and historic_mode != "enqueue":
        raise ValueError('force=True is only valid with historic_mode="enqueue"')

    # Mode 1: data dict provided — legacy mode
    if data is not None:
        return call_validation(client, "orchestrator", data, wait=wait, external_id=external_id)

    from pathlib import Path

    import yaml

    from cognite_data_quality.deploy import DeploymentSettings, ViewConfig, load_settings

    mode4_view_only = (
        data_product_external_id is None
        and view_external_id is not None
        and settings_path is None
        and view_config_external_id is None
    )

    # Mode 4: DataProduct + DMS — orchestrator or historic queue enqueue
    if data_product_external_id is not None or mode4_view_only:
        if historic_mode == "enqueue":
            return enqueue_historic_validation(
                client,
                data_product_external_id=data_product_external_id,
                data_product_version=data_product_version,
                view_external_id=view_external_id,
                view_space=view_space,
                data_quality_space=data_quality_space,
                trigger_queue_manager=trigger_queue_manager,
                queue_manager_workflow_external_id=queue_manager_workflow_external_id,
                queue_manager_workflow_version=queue_manager_workflow_version,
                force=force,
            ).to_dict()

        settings_obj, data_product_external_id, _resolved_version, view_configs = _load_mode4_view_configs(
            client,
            data_quality_space=data_quality_space,
            data_product_external_id=data_product_external_id,
            data_product_version=data_product_version,
            view_external_id=view_external_id,
            view_space=view_space,
        )

        from cognite_data_quality._historic_queue_types import OrchestratorTriggerResult

        results: list[OrchestratorTriggerResult] = []
        for vc in view_configs:
            orchestrator_data = _build_orchestrator_data(
                settings_obj,
                vc,
                dataproduct_external_id=data_product_external_id,
            )
            print(f"Triggering {vc.view.space}/{vc.view.external_id}...")
            payload = {"validation_type": "orchestrator", **orchestrator_data}
            fn_call = client.functions.call(
                external_id=settings_obj.function.external_id,
                data=payload,
                wait=False,
            )
            call_id = fn_call.id
            results.append(
                OrchestratorTriggerResult(
                    view=vc.view.external_id,
                    view_space=vc.view.space,
                    data_product=data_product_external_id,
                    call_id=call_id,
                    status="Running",
                )
            )
            print(f"  Triggered {vc.view.space}/{vc.view.external_id}: call_id={call_id}")
        return [r.to_dict() for r in results]

    # Mode 3: fetch both view config and settings from CDF Files — no local files needed
    if view_config_external_id is not None:
        settings_obj = (
            load_settings(Path(settings_path))
            if settings_path is not None
            else DeploymentSettings(
                **yaml.safe_load(client.files.download_bytes(external_id=settings_external_id).decode("utf-8")) or {}
            )
        )
        raw_view = client.files.download_bytes(external_id=view_config_external_id)
        view_config = ViewConfig(**yaml.safe_load(raw_view.decode("utf-8")) or {})
        data = _build_orchestrator_data(settings_obj, view_config)
        return call_validation(client, "orchestrator", data, wait=wait, external_id=settings_obj.function.external_id)

    # Mode 2: scan local YAML files
    if settings_path is None:
        raise ValueError("Must provide either 'data' dict, 'view_config_external_id', or 'settings_path'")

    if view_external_id is None:
        raise ValueError("Must provide 'view_external_id' or 'view_config_external_id' when using settings_path mode")

    settings_obj = load_settings(Path(settings_path))
    base_dir = Path(settings_path).parent
    if views_dir is None:
        views_dir_obj = base_dir / "views"
    else:
        views_dir_obj = Path(views_dir) if not isinstance(views_dir, Path) else views_dir

    view_config = _find_view_config_in_dir(views_dir_obj, view_external_id, view_space)

    data = _build_orchestrator_data(settings_obj, view_config)
    return call_validation(client, "orchestrator", data, wait=wait, external_id=settings_obj.function.external_id)

enqueue_historic_validation(client, *, data_product_external_id=None, data_product_version=None, view_external_id=None, view_space=None, data_quality_space='dataQuality', trigger_queue_manager=False, queue_manager_workflow_external_id='dq-historic-queue-manager', queue_manager_workflow_version='1', force=False)

Enqueue historic validation jobs on HistoricJobQueue for sequential processing.

Idempotent per view and DataProduct version: jobs already queued, running, or completed for the same version are skipped. Failed jobs are re-enqueued.

Pass force=True to re-queue a completed historic run for the current DataProduct/RuleSet version, including views with historic_reprocess_on_update=false. Already queued or running jobs are still skipped. view_external_id only narrows which views are in scope.

All managed DataProducts (from DMS managed_data_product_external_ids):

enqueue_historic_validation(client, data_quality_space="sp_data_quality")

Single DataProduct (all ruled views, or filter with view_external_id):

enqueue_historic_validation(
    client,
    data_product_external_id="dp_ent_sif",
    data_quality_space="sp_data_quality",
)

Re-run historic for all ruled views without a version bump:

enqueue_historic_validation(
    client,
    data_product_external_id="dp_ent_sif",
    data_quality_space="sp_data_quality",
    force=True,
    trigger_queue_manager=True,
)

Parameters:

Name	Type	Description	Default
client	CogniteClient	Cognite client instance.	required
data_product_external_id	str | None	DataProduct external ID. When omitted, enqueues for every managed DataProduct in DMS settings.	None
data_product_version	str | None	Optional explicit DP version; defaults to latest published.	None
view_external_id	str | list[str] | None	Optional view filter (single ID or list).	None
view_space	str | None	Disambiguate when the same view external ID exists in multiple spaces.	None
data_quality_space	str	DMS space for DataQualitySettings and HistoricJobQueue.	'dataQuality'
trigger_queue_manager	bool	If True, start dq-historic-queue-manager once after enqueue.	False
queue_manager_workflow_external_id	str	Workflow external ID for the queue manager.	'dq-historic-queue-manager'
queue_manager_workflow_version	str	Workflow version for the queue manager.	'1'
force	bool	If True, re-queue completed jobs for the same DataProduct/RuleSet version and include views with historic_reprocess_on_update=false. Does not redeploy workflows (that is data_product_sync force: true).	False

Returns:

Type	Description
EnqueueHistoricValidationResult	EnqueueHistoricValidationResult (use .to_dict() for JSON-serializable dict).

Source code in cognite_data_quality/invoke.py

def enqueue_historic_validation(
    client: CogniteClient,
    *,
    data_product_external_id: str | None = None,
    data_product_version: str | None = None,
    view_external_id: str | list[str] | None = None,
    view_space: str | None = None,
    data_quality_space: str = "dataQuality",
    trigger_queue_manager: bool = False,
    queue_manager_workflow_external_id: str = "dq-historic-queue-manager",
    queue_manager_workflow_version: str = "1",
    force: bool = False,
) -> EnqueueHistoricValidationResult:
    """Enqueue historic validation jobs on ``HistoricJobQueue`` for sequential processing.

    Idempotent per view and DataProduct version: jobs already ``queued``, ``running``,
    or ``completed`` for the same version are skipped. Failed jobs are re-enqueued.

    Pass ``force=True`` to re-queue a completed historic run for the current
    DataProduct/RuleSet version, including views with
    ``historic_reprocess_on_update=false``. Already ``queued`` or ``running`` jobs
    are still skipped. ``view_external_id`` only narrows which views are in scope.

    **All managed DataProducts** (from DMS ``managed_data_product_external_ids``):

        enqueue_historic_validation(client, data_quality_space="sp_data_quality")

    **Single DataProduct** (all ruled views, or filter with ``view_external_id``):

        enqueue_historic_validation(
            client,
            data_product_external_id="dp_ent_sif",
            data_quality_space="sp_data_quality",
        )

    **Re-run historic for all ruled views without a version bump:**

        enqueue_historic_validation(
            client,
            data_product_external_id="dp_ent_sif",
            data_quality_space="sp_data_quality",
            force=True,
            trigger_queue_manager=True,
        )

    Args:
        client: Cognite client instance.
        data_product_external_id: DataProduct external ID. When omitted, enqueues for
            every managed DataProduct in DMS settings.
        data_product_version: Optional explicit DP version; defaults to latest published.
        view_external_id: Optional view filter (single ID or list).
        view_space: Disambiguate when the same view external ID exists in multiple spaces.
        data_quality_space: DMS space for ``DataQualitySettings`` and ``HistoricJobQueue``.
        trigger_queue_manager: If True, start ``dq-historic-queue-manager`` once after enqueue.
        queue_manager_workflow_external_id: Workflow external ID for the queue manager.
        queue_manager_workflow_version: Workflow version for the queue manager.
        force: If True, re-queue completed jobs for the same DataProduct/RuleSet version
            and include views with ``historic_reprocess_on_update=false``. Does not
            redeploy workflows (that is ``data_product_sync`` ``force: true``).

    Returns:
        ``EnqueueHistoricValidationResult`` (use ``.to_dict()`` for JSON-serializable dict).
    """
    from cognite_data_quality._containers import decode_managed_dp_ids
    from cognite_data_quality._historic_queue import enqueue_historic_jobs, trigger_historic_queue_manager
    from cognite_data_quality._historic_queue_types import EnqueueHistoricValidationResult

    dms_props, _settings_obj = _load_dms_settings_or_raise(client, data_quality_space)

    dp_targets: list[tuple[str, str | None]] = []
    if data_product_external_id is not None:
        dp_targets.append((data_product_external_id, data_product_version))
    else:
        if view_external_id is not None:
            _settings_obj, dp_id, resolved_version, view_configs = _load_mode4_view_configs(
                client,
                data_quality_space=data_quality_space,
                data_product_external_id=None,
                data_product_version=data_product_version,
                view_external_id=view_external_id,
                view_space=view_space,
            )
            result = enqueue_historic_jobs(
                client,
                data_quality_space,
                view_configs,
                dp_id,
                resolved_version,
                force=force,
            )
            queue_execution_id = None
            if trigger_queue_manager:
                queue_execution_id = trigger_historic_queue_manager(
                    client,
                    workflow_external_id=queue_manager_workflow_external_id,
                    version=queue_manager_workflow_version,
                )
            return EnqueueHistoricValidationResult(
                data_products=[result],
                total_enqueued=result.enqueued,
                total_skipped=result.skipped,
                warnings=list(result.warnings),
                queue_manager_execution_id=queue_execution_id,
            )

        managed_ids = decode_managed_dp_ids(dms_props.get("managed_data_product_external_ids")) or []
        if not managed_ids:
            raise ValueError(
                "No managed_data_product_external_ids in DMS settings. "
                "Pass data_product_external_id or run deploy_validation_infrastructure() first."
            )
        dp_targets = [(dp_id, None) for dp_id in managed_ids]

    data_products: list[Any] = []
    total_enqueued = 0
    total_skipped = 0
    all_warnings: list[str] = []

    for dp_id, dp_version in dp_targets:
        _settings_obj, resolved_dp_id, resolved_version, view_configs = _load_mode4_view_configs(
            client,
            data_quality_space=data_quality_space,
            data_product_external_id=dp_id,
            data_product_version=dp_version,
            view_external_id=view_external_id,
            view_space=view_space,
            allow_empty=data_product_external_id is None,
        )
        if not view_configs:
            warning = (
                f"No views with SHACL rules in {resolved_dp_id}@{resolved_version}; "
                "skipping historic enqueue for this DataProduct"
            )
            print(f"  WARNING: {warning}")
            all_warnings.append(warning)
            continue
        print(f"Enqueueing historic jobs for {resolved_dp_id}@{resolved_version} ({len(view_configs)} view(s))...")
        result = enqueue_historic_jobs(
            client,
            data_quality_space,
            view_configs,
            resolved_dp_id,
            resolved_version,
            force=force,
        )
        data_products.append(result)
        total_enqueued += result.enqueued
        total_skipped += result.skipped
        all_warnings.extend(result.warnings)

    queue_execution_id = None
    if trigger_queue_manager:
        queue_execution_id = trigger_historic_queue_manager(
            client,
            workflow_external_id=queue_manager_workflow_external_id,
            version=queue_manager_workflow_version,
        )

    return EnqueueHistoricValidationResult(
        data_products=data_products,
        total_enqueued=total_enqueued,
        total_skipped=total_skipped,
        warnings=all_warnings,
        queue_manager_execution_id=queue_execution_id,
    )