Airflow Data Quality Provider part 1

dev/breeze/tests/test_selective_checks.py

@@ -2884,7 +2884,7 @@ def test_upgrade_to_newer_dependencies(
             ("providers/common/sql/src/airflow/providers/common/sql/common_sql_python.py",),
             {
                 "docs-list-as-string": "amazon apache.drill apache.druid apache.hive apache.iceberg "
-                "apache.impala apache.pinot clickhousedb common.ai common.compat common.sql databricks elasticsearch "
+                "apache.impala apache.pinot clickhousedb common.ai common.compat common.dataquality common.sql databricks elasticsearch "
                 "exasol google informatica jdbc microsoft.mssql mysql odbc openlineage "
                 "oracle pgvector postgres presto slack snowflake sqlite teradata trino vertica ydb",
             },

docs/spelling_wordlist.txt

@@ -294,6 +294,7 @@ conda
 conf
 Config
 config
+ConfigDict
 configfile
 configMap
 configmap

providers/common/dataquality/.pre-commit-config.yaml

@@ -0,0 +1,36 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+---
+default_stages: [pre-commit, pre-push]
+minimum_prek_version: '0.3.4'
+default_language_version:
+  python: python3
+  node: 22.19.0
+  golang: 1.24.0
+repos:
+  - repo: local
+    hooks:
+      - id: generate-common-dataquality-ruleset-schema
+        name: Generate Data Quality RuleSet schema
+        language: python
+        entry: ../../../scripts/ci/prek/generate_common_dataquality_ruleset_schema.py
+        pass_filenames: false
+        always_run: true
+        files: >
+          (?x)
+          ^src/airflow/providers/common/dataquality/rules/.*\.py$|
+          ^src/airflow/providers/common/dataquality/skills/dataquality-rule-authoring/references/ruleset\.schema\.json$

providers/common/dataquality/README.rst

@@ -19,7 +19,11 @@ Package ``apache-airflow-providers-common-dataquality``
 
 Release: ``0.1.0``
 
-Common Data Quality Provider
+``Data Quality Provider``
+
+Declarative data quality rules with durable, per-rule execution history. Checks run through
+``common.sql`` DB-API hooks; results are persisted to a configurable results store (object
+storage or local files) so task, run, and rule-level quality can be inspected over time.
 
 Provider package
 ----------------
@@ -37,8 +41,12 @@ see ``Requirements`` below.
 Requirements
 ------------
 
-==================  ==================
-PIP package         Version required
-==================  ==================
-``apache-airflow``  ``>=3.0.0``
-==================  ==================
+==========================================  ==================
+PIP package                                 Version required
+==========================================  ==================
+``apache-airflow``                          ``>=3.0.0``
+``apache-airflow-providers-common-compat``  ``>=1.15.0``
+``apache-airflow-providers-common-sql``     ``>=2.0.0``
+``pydantic``                                ``>=2.11.0``
+``pyyaml``                                  ``>=6.0.2``
+==========================================  ==================

providers/common/dataquality/docs/agents.rst

@@ -0,0 +1,55 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+ ..   http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+.. _dq:agents:
+
+Generating rules with an LLM
+==============================
+
+Writing a :class:`~airflow.providers.common.dataquality.rules.RuleSet` by hand for every table doesn't scale.
+An LLM can propose one from a table's column definitions instead, given the check catalog as
+context.
+
+The ``dataquality-rule-authoring`` skill
+----------------------------------------
+
+This provider ships an `Agent Skill <https://agentskills.io>`__ at
+``airflow/providers/common/dataquality/skills/dataquality-rule-authoring/``: a ``SKILL.md`` documenting the
+``RuleSet``/``DQRule`` fields and check catalog, plus a generated JSON Schema
+(``references/ruleset.schema.json``) for validation.
+
+Point ``common.ai``'s :doc:`AgentSkillsToolset <apache-airflow-providers-common-ai:toolsets>` at
+it, and give the model ``output_type=RuleSet`` so pydantic-ai validates -- and self-corrects --
+its output before the task completes:
+
+.. exampleinclude:: /../src/airflow/providers/common/dataquality/example_dags/example_dq_llm_generated_ruleset.py
+    :language: python
+    :start-after: [START howto_task_dq_generate_ruleset_with_llm]
+    :end-before: [END howto_task_dq_generate_ruleset_with_llm]
+
+Requires ``apache-airflow-providers-common-ai[skills]`` and a configured ``llm_conn_id``.
+
+Wiring the result into a check
+---------------------------------
+
+``@task.dq_check`` can leave ``ruleset=`` unset and return the LLM task's result at execution
+time instead:
+
+.. exampleinclude:: /../src/airflow/providers/common/dataquality/example_dags/example_dq_llm_generated_ruleset.py
+    :language: python
+    :start-after: [START howto_decorator_dq_check_llm_runtime_ruleset]
+    :end-before: [END howto_decorator_dq_check_llm_runtime_ruleset]

providers/common/dataquality/docs/assets.rst

@@ -0,0 +1,70 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+ ..   http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+.. _dq:assets:
+
+Assets and quality gating
+============================
+
+Quality rules can travel with the :class:`~airflow.sdk.Asset` they describe instead of being
+scattered across every Dag that checks it, and a downstream consumer Dag can refuse to run when
+the data it was triggered by did not meet a minimum quality bar.
+
+Attaching a ruleset to an asset
+----------------------------------
+
+:func:`~airflow.providers.common.dataquality.assets.asset_quality` stores ruleset, connection, and table
+configuration inside ``Asset.extra`` under the ``airflow.dataquality`` key, so it is serialized with the
+Dag and needs no Airflow core changes:
+
+.. exampleinclude:: /../src/airflow/providers/common/dataquality/example_dags/example_dq_require_quality.py
+    :language: python
+    :start-after: [START howto_asset_quality]
+    :end-before: [END howto_asset_quality]
+
+Pass the resulting asset to :class:`~airflow.providers.common.dataquality.operators.dq_check.DQCheckOperator`
+via ``asset=`` (see :doc:`operators`) instead of ``table``/``ruleset``/``conn_id``: the operator
+resolves all three from the asset's config, adds the asset to its own outlets, and attaches its
+summary -- including the quality ``score`` used below -- to the asset event.
+
+Only one Dag should call ``asset_quality()`` for a given asset ``name``/``uri``. Airflow keeps one
+shared record per asset across all Dags, so if more than one Dag attaches (or omits) config for the
+same asset, whichever Dag parsed most recently determines what's stored.
+
+Gating a consumer Dag on quality
+------------------------------------
+
+:func:`~airflow.providers.common.dataquality.assets.require_quality` builds a ``@task.short_circuit`` task that
+reads the ``score`` off the asset event that triggered the current run, and skips every
+downstream task when that event has no quality summary at all, or its score is below
+``min_score``:
+
+.. exampleinclude:: /../src/airflow/providers/common/dataquality/example_dags/example_dq_require_quality.py
+    :language: python
+    :start-after: [START howto_require_quality]
+    :end-before: [END howto_require_quality]
+
+Put the gate first in a Dag scheduled by the asset, and chain everything else after it:
+
+.. code-block:: python
+
+    with DAG("orders_consumer", schedule=orders_asset) as consumer:
+        gate = require_quality(orders_asset, min_score=0.95)
+        gate >> process_orders()
+
+``min_score`` must be between ``0`` and ``1``; the check considers only the *most recent*
+triggering event for the asset when a run was triggered by several.

providers/common/dataquality/docs/configurations-ref.rst

@@ -0,0 +1,19 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+ ..   http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+.. include:: /../../../../devel-common/src/sphinx_exts/includes/providers-configurations-ref.rst
+.. include:: /../../../../devel-common/src/sphinx_exts/includes/sections-and-options.rst

providers/common/dataquality/docs/decorators.rst

@@ -0,0 +1,52 @@
+ .. Licensed to the Apache Software Foundation (ASF) under one
+    or more contributor license agreements.  See the NOTICE file
+    distributed with this work for additional information
+    regarding copyright ownership.  The ASF licenses this file
+    to you under the Apache License, Version 2.0 (the
+    "License"); you may not use this file except in compliance
+    with the License.  You may obtain a copy of the License at
+
+ ..   http://www.apache.org/licenses/LICENSE-2.0
+
+ .. Unless required by applicable law or agreed to in writing,
+    software distributed under the License is distributed on an
+    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+    KIND, either express or implied.  See the License for the
+    specific language governing permissions and limitations
+    under the License.
+
+.. _howto/decorator:dq_check:
+
+``@task.dq_check``
+=====================
+
+``@task.dq_check`` wraps :class:`~airflow.providers.common.dataquality.operators.dq_check.DQCheckOperator` in
+the TaskFlow API. ``ruleset`` may be declared as a decorator argument when it exists at
+Dag-parse time, or returned by the decorated function as a runtime ruleset. ``table`` or
+``asset`` are declared as decorator arguments exactly like the plain operator. The decorated
+function is optional plumbing on top: return ``None`` to run the check exactly as declared.
+
+.. exampleinclude:: /../tests/system/common/dataquality/example_dq_check.py
+    :language: python
+    :dedent: 4
+    :start-after: [START howto_decorator_dq_check]
+    :end-before: [END howto_decorator_dq_check]
+
+Runtime rule sets
+--------------------
+
+Return a :class:`~airflow.providers.common.dataquality.rules.RuleSet`, its dict form, or a YAML path to use a
+ruleset that is only known at task-execution time -- for example one produced by an upstream
+task, loaded from a Variable, or generated by an LLM. Return ``None`` to use the ruleset declared
+on the decorator.
+
+Swapping in a different ruleset at execution time:
+
+.. exampleinclude:: /../src/airflow/providers/common/dataquality/example_dags/example_dq_check_decorator_dynamic.py
+    :language: python
+    :start-after: [START howto_decorator_dq_check_runtime_ruleset]
+    :end-before: [END howto_decorator_dq_check_runtime_ruleset]
+
+Everything documented for the plain operator in :doc:`operators` -- ``fail_on``, persistence,
+``asset``, ``partition_clause``, ``custom_sql`` -- applies unchanged; the decorator only adds the
+optional runtime ruleset step before the check runs.