pierrejeambrun commented on code in PR #68785:
URL: https://github.com/apache/airflow/pull/68785#discussion_r3506895915
##########
airflow-core/src/airflow/api_fastapi/common/parameters.py:
##########
@@ -872,6 +851,31 @@ def depends(
) -> _TagsFilter:
return cls().set_value(_TagFilterModel(tags=tags,
tags_match_mode=tags_match_mode))
+class _DagTagFilterModel(BaseModel):
+ """Tag Filter Model with a match mode parameter."""
+
+ tags: list[str]
+
+class _DagTagsFilter(BaseParam[_DagTagFilterModel]):
+
+ def to_orm(self, select: Select) -> Select:
+ if self.skip_none is False:
+ raise ValueError(f"Cannot set 'skip_none' to False on a
{type(self)}")
+
+ if not self.value or not self.value.tags:
+ return select
+
+ conditions = [DagModel.tags.any(DagTag.name == tag) for tag in
self.value.tags]
+ return select.where(*conditions, DagRun.dag_id == DagModel.dag_id )
+
+ @classmethod
+ def depends(
+ cls,
+ tags: list[str] = Query(default_factory=list)
+ ) -> _DagTagsFilter:
+ return cls().set_value(_DagTagFilterModel(tags=tags))
+
+
Review Comment:
Please re-use `_TagFilterModel` and `_TagsFilter`. This is almost a perfect
copy.
##########
airflow-core/src/airflow/migrations/versions/a1f939e903c3_add_new_field_to_db.py:
##########
@@ -16,39 +16,38 @@
# specific language governing permissions and limitations
# under the License.
-"""
-Add partition_date to asset_partition_dag_run.
-
-The target datetime is frozen at APDR creation time so the consumer DagRun's
-``partition_date`` is consistent with the partition mapper that produced its
-``partition_key``.
+"""add new field to db
-Revision ID: d2f4e1b3c5a7
+Revision ID: a1f939e903c3
Revises: 9ff64e1c35d3
-Create Date: 2026-05-21 09:00:00.000000
-"""
+Create Date: 2026-06-26 04:45:59.320339
-from __future__ import annotations
+"""
import sqlalchemy as sa
from alembic import op
-from airflow.utils.sqlalchemy import UtcDateTime
-revision = "d2f4e1b3c5a7"
+# revision identifiers, used by Alembic.
+revision = "a1f939e903c3"
down_revision = "9ff64e1c35d3"
branch_labels = None
depends_on = None
-airflow_version = "3.3.0"
def upgrade():
- """Add partition_date column to asset_partition_dag_run."""
- with op.batch_alter_table("asset_partition_dag_run", schema=None) as
batch_op:
- batch_op.add_column(sa.Column("partition_date", UtcDateTime,
nullable=True))
+ """Apply add new field to db"""
+ # ### commands auto generated by Alembic - please adjust! ###
+ with op.batch_alter_table('dag_run', schema=None) as batch_op:
+ batch_op.add_column(sa.Column('tags', sa.String(length=512),
nullable=True))
+
+ # ### end Alembic commands ###
def downgrade():
- """Remove partition_date column from asset_partition_dag_run."""
- with op.batch_alter_table("asset_partition_dag_run", schema=None) as
batch_op:
- batch_op.drop_column("partition_date")
+ """Unapply add new field to db"""
+ # ### commands auto generated by Alembic - please adjust! ###
+ with op.batch_alter_table('dag_run', schema=None) as batch_op:
+ batch_op.drop_column('tags')
+
+ # ### end Alembic commands ###
Review Comment:
Misunderstanding of the related issue. We are not adding a `tags` column to
dag runs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
