Quality Checks

The Quality resource (client.quality) lets you manage quality checks, trigger executions, push external results, and retrieve statistics.

Methods

list(space_slug, *, include_archived=)

List all quality checks in a space.

Parameter	Type	Default	Description
space_slug	str	required	Space identifier
include_archived	bool	False	Include archived checks

Returns: list[QualityCheck]

checks = await client.quality.list("marketing-analytics")

---

get(space_slug, check_slug)

Get a quality check by slug.

Parameter	Type	Description
space_slug	str	Space identifier
check_slug	str	Check slug

Returns: QualityCheck

check = await client.quality.get("marketing-analytics", "row-count-check")

---

get_by_id(check_id)

Get a quality check by UUID.

Returns: QualityCheck

---

create(space_slug, *, name, check_type, query=, product_ids=, connector_id=, description=, parameters=, **extra)

Create a new quality check.

Parameter	Type	Default	Description
space_slug	str	required	Target space
name	str	required	Check name
check_type	str	required	Type (sql, row_count, manual, etc.)
query	str | None	None	SQL query
product_ids	list[UUID] | None	None	Linked product UUIDs
connector_id	UUID | None	None	Connector to execute against
description	str | None	None	Description
parameters	dict | None	None	Parameter definitions for query placeholders

Returns: QualityCheck

check = await client.quality.create(
    "marketing-analytics",
    name="Row Count Check",
    check_type="sql",
    query="SELECT COUNT(*) FROM events WHERE date = '{{run_date}}'",
    parameters={"run_date": "2024-01-15"},
    connector_id=connector_id,
    product_ids=[product_id],
)

---

trigger(space_slug, check_slug, *, manual_value=, parameters=)

Trigger execution of a quality check.

Parameter	Type	Default	Description
space_slug	str	required	Space identifier
check_slug	str	required	Check slug
manual_value	float | None	None	Value for manual checks
parameters	dict | None	None	Runtime parameter overrides

Returns: CheckExecution

result = await client.quality.trigger(
    "marketing-analytics",
    "row-count-check",
    parameters={"run_date": "2024-01-15"},
)
print(f"Passed: {result.is_passed}")

---

push_result(space_slug, check_slug, *, status, value=, metadata=, **extra)

Push an external execution result (e.g., from Airflow or CI pipelines).

Parameter	Type	Default	Description
space_slug	str	required	Space identifier
check_slug	str	required	Check slug
status	str	required	Result status (passed, failed, warning)
value	float | None	None	Measured value
metadata	dict | None	None	Additional metadata

Returns: CheckExecution

execution = await client.quality.push_result(
    "marketing-analytics",
    "freshness-check",
    status="passed",
    value=42.0,
)

---

get_executions(space_slug, check_slug)

Get execution history for a check.

Returns: list[CheckExecution]

---

get_stats(check_id)

Get aggregated statistics for a check by UUID.

Returns: CheckStats

---

register_external_check(space_slug, check_slug, *, status, value=, name=, description=, product_ids=, metadata=, **extra)

Upsert an external check — creates it if it does not exist, then pushes the result. This is the recommended one-call method for CI/CD integrations.

Parameter	Type	Default	Description
space_slug	str	required	Space identifier
check_slug	str	required	Check slug
status	str	required	Result status
value	float | None	None	Measured value
name	str | None	None	Check name (used on creation)
description	str | None	None	Check description
product_ids	list[UUID] | None	None	Linked products
metadata	dict | None	None	Extra metadata

Returns: CheckExecution

execution = await client.quality.register_external_check(
    "marketing-analytics",
    "airflow-freshness",
    status="passed",
    value=3.5,
    name="Airflow Freshness Check",
    product_ids=[product_id],
)

---

list_for_product(space_slug, product_slug)

List all quality checks linked to a product.

Returns: list[QualityCheck]

---

get_product_trends(space_slug, product_id)

Get daily quality trend aggregation for a product.

Returns: trend data dict

---

Configuration-Driven Methods

These methods let you manage quality checks as code using YAML configuration files. See the DQ Config reference for the YAML format.

validate_config(config)

Validate a parsed config without making any changes. Returns a list of human-readable errors (empty on success).

Parameter	Type	Description
config	DqConfig	Parsed configuration

Returns: list[str]

from qarion.models.dq_config import DqConfig

config = DqConfig.from_yaml("qarion-dq.yaml")
errors = await client.quality.validate_config(config)
if errors:
    for err in errors:
        print(f"Error: {err}")

---

apply_config(config)

Sync quality check definitions from the config to the platform. Creates missing checks and updates existing ones.

Parameter	Type	Description
config	DqConfig	Parsed configuration

Returns: dict — {"created": [...], "updated": [...], "unchanged": [...]}

summary = await client.quality.apply_config(config)
print(f"Created: {summary['created']}")
print(f"Updated: {summary['updated']}")

---

run_config(config, *, record_results=True)

Execute all checks defined in the config.

Parameter	Type	Default	Description
config	DqConfig	required	Parsed configuration
record_results	bool	True	Record results on the platform

Returns: list[CheckExecution]

results = await client.quality.run_config(config)
for r in results:
    print(f"{r.status}: {r.value}")