This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 84f75158120 Require `common.sql` 2.2.0 for common.ai's SQL extras
(#73867)
84f75158120 is described below
commit 84f751581201739bfaa504854200c7807359b5d8
Author: Kaxil Naik <[email protected]>
AuthorDate: Mon Sep 28 23:38:19 2026 +0100
Require `common.sql` 2.2.0 for common.ai's SQL extras (#73867)
The LLMSchemaCompareOperator guide and example now show a DataSourceConfig
holding only conn_id and table_name. That form needs common.sql 2.2.0; on
2.1.1 it raises "Unsupported storage type for URI". Raise the floor of the
sql and common.sql extras to match.
Also fix two docs issues found while testing 0.10.0rc1: comments in the
classifier example Dag that predate decision_policy and the decision XCom,
and a categories={...} placeholder in the retry policy guide that raised
TypeError when copied.
---
providers/common/ai/README.rst | 4 +--
providers/common/ai/docs/index.rst | 4 +--
providers/common/ai/docs/retry_policies.rst | 39 +++++++++++-----------
providers/common/ai/pyproject.toml | 4 +--
.../ai/example_dags/example_classifier_model.py | 8 +++--
5 files changed, 30 insertions(+), 29 deletions(-)
diff --git a/providers/common/ai/README.rst b/providers/common/ai/README.rst
index b4b86e5f091..9d196fafb8d 100644
--- a/providers/common/ai/README.rst
+++ b/providers/common/ai/README.rst
@@ -98,8 +98,8 @@ Extra Dependencies
``skills`` ``apache-airflow-providers-git>=0.4.0``,
``pydantic-ai-skills>=1.2.0``
``avro`` ``fastavro>=1.10.0; python_version < "3.14"``,
``fastavro>=1.12.1; python_version >= "3.14"``
``parquet`` ``pyarrow>=18.0.0; python_version < '3.14'``,
``pyarrow>=22.0.0; python_version >= '3.14'``
-``sql`` ``apache-airflow-providers-common-sql>=1.33.0``,
``sqlglot>=30.0.0``
-``common.sql`` ``apache-airflow-providers-common-sql>=1.33.0``
+``sql`` ``apache-airflow-providers-common-sql>=2.2.0``,
``sqlglot>=30.0.0``
+``common.sql`` ``apache-airflow-providers-common-sql>=2.2.0``
``langchain`` ``langchain>=1.0.0``
``llamaindex`` ``dataclasses-json>=0.6.7``, ``llama-index-core>=0.14.5``,
``llama-index-embeddings-openai>=0.6.0``, ``llama-index-llms-openai>=0.6.8``
``pdf`` ``pypdf>=4.0.0``
diff --git a/providers/common/ai/docs/index.rst
b/providers/common/ai/docs/index.rst
index ba37e88a646..2ac36424068 100644
--- a/providers/common/ai/docs/index.rst
+++ b/providers/common/ai/docs/index.rst
@@ -229,8 +229,8 @@ Extra Dependencies
``skills`` ``apache-airflow-providers-git>=0.4.0``,
``pydantic-ai-skills>=1.2.0``
``avro`` ``fastavro>=1.10.0; python_version < "3.14"``,
``fastavro>=1.12.1; python_version >= "3.14"``
``parquet`` ``pyarrow>=18.0.0; python_version < '3.14'``,
``pyarrow>=22.0.0; python_version >= '3.14'``
-``sql`` ``apache-airflow-providers-common-sql>=1.33.0``,
``sqlglot>=30.0.0``
-``common.sql`` ``apache-airflow-providers-common-sql>=1.33.0``
+``sql`` ``apache-airflow-providers-common-sql>=2.2.0``,
``sqlglot>=30.0.0``
+``common.sql`` ``apache-airflow-providers-common-sql>=2.2.0``
``langchain`` ``langchain>=1.0.0``
``llamaindex`` ``dataclasses-json>=0.6.7``, ``llama-index-core>=0.14.5``,
``llama-index-embeddings-openai>=0.6.0``, ``llama-index-llms-openai>=0.6.8``
``pdf`` ``pypdf>=4.0.0``
diff --git a/providers/common/ai/docs/retry_policies.rst
b/providers/common/ai/docs/retry_policies.rst
index cd88962473f..a94f839056b 100644
--- a/providers/common/ai/docs/retry_policies.rst
+++ b/providers/common/ai/docs/retry_policies.rst
@@ -234,25 +234,24 @@ on them differently:
.. code-block:: python
- snowflake_policy = ClassifierRetryPolicy(
- llm_conn_id="pydanticai_default",
- categories={
- "queued": ErrorCategory(
- "Statement queued or a concurrency limit reached; the
warehouse is busy.",
- delay=timedelta(seconds=120),
- ),
- "warehouse_suspended": ErrorCategory(
- "The warehouse is suspended and will auto-resume.",
delay=timedelta(seconds=30)
- ),
- "token_expired": ErrorCategory(
- "A JWT or session token expired; the token rotates on its
own.", delay=timedelta(seconds=30)
- ),
- "schema_drift": ErrorCategory(
- "A referenced column, table or view does not exist; a person
has to fix the schema.",
- retry=False,
- ),
- },
- )
+ SNOWFLAKE_CATEGORIES = {
+ "queued": ErrorCategory(
+ "Statement queued or a concurrency limit reached; the warehouse is
busy.",
+ delay=timedelta(seconds=120),
+ ),
+ "warehouse_suspended": ErrorCategory(
+ "The warehouse is suspended and will auto-resume.",
delay=timedelta(seconds=30)
+ ),
+ "token_expired": ErrorCategory(
+ "A JWT or session token expired; the token rotates on its own.",
delay=timedelta(seconds=30)
+ ),
+ "schema_drift": ErrorCategory(
+ "A referenced column, table or view does not exist; a person has
to fix the schema.",
+ retry=False,
+ ),
+ }
+
+ snowflake_policy = ClassifierRetryPolicy(llm_conn_id="pydanticai_default",
categories=SNOWFLAKE_CATEGORIES)
Write descriptions as the boundary between categories: what belongs here and
what
does not. That is the whole of what the model reads about a category; the name
@@ -474,7 +473,7 @@ not decide the action:
snowflake_policy = ClassifierRetryPolicy(
llm_conn_id="pydanticai_default",
instructions=SNOWFLAKE_HINTS,
- categories={...}, # the same names the hints use
+ categories=SNOWFLAKE_CATEGORIES, # the table above; the hints use the
same names
fallback_rules=[
RetryRule(
exception=ConnectionError,
diff --git a/providers/common/ai/pyproject.toml
b/providers/common/ai/pyproject.toml
index 7cb3c823e35..dd818c99ad9 100644
--- a/providers/common/ai/pyproject.toml
+++ b/providers/common/ai/pyproject.toml
@@ -128,11 +128,11 @@ dependencies = [
"pyarrow>=22.0.0; python_version >= '3.14'",
]
"sql" = [
- "apache-airflow-providers-common-sql>=1.33.0",
+ "apache-airflow-providers-common-sql>=2.2.0",
"sqlglot>=30.0.0",
]
"common.sql" = [
- "apache-airflow-providers-common-sql>=1.33.0"
+ "apache-airflow-providers-common-sql>=2.2.0"
]
"langchain" = [
"langchain>=1.0.0",
diff --git
a/providers/common/ai/src/airflow/providers/common/ai/example_dags/example_classifier_model.py
b/providers/common/ai/src/airflow/providers/common/ai/example_dags/example_classifier_model.py
index c108a3cacae..b56124f50fc 100644
---
a/providers/common/ai/src/airflow/providers/common/ai/example_dags/example_classifier_model.py
+++
b/providers/common/ai/src/airflow/providers/common/ai/example_dags/example_classifier_model.py
@@ -96,8 +96,9 @@ example_classifier_model_branch()
def example_classifier_model_confidence():
"""Classify a failure and escalate when the model says it does not know.
- The branch Dag above cannot do this: ``LLMBranchOperator`` takes the
branch inside the
- operator, before any task can read the confidence.
+ ``LLMBranchOperator`` can gate its pick on one confidence bar through
``decision_policy``.
+ Reading the confidence in a task of your own, as here, lets you act
differently in more
+ than one band.
"""
@task
@@ -110,7 +111,8 @@ def example_classifier_model_confidence():
# Confidence is reported per output field; a bare output type lands
under
# "response". A bounded ``float`` output would report none at all --
there the
# probability is the answer -- so this ``or 0.0`` would read as no
confidence
- # rather than as a missing one. No operator surfaces this on XCom by
default.
+ # rather than as a missing one. ``LLMOperator`` and
``LLMBranchOperator`` push the
+ # same confidence in their ``decision`` XCom.
details = result.response.provider_details or {}
return {
"category": result.output,
