[GitHub] [airflow] ashb commented on a change in pull request #9204: Add filtering tasks/task instances by tags

[GitBox <gi...@apache.org>]

ashb commented on a change in pull request #9204:
URL: https://github.com/apache/airflow/pull/9204#discussion_r440077581



##########
File path: airflow/models/dag.py
##########
@@ -844,11 +844,12 @@ def get_task_instances(
             start_date = timezone.make_aware(
                 datetime.combine(start_date, datetime.min.time()))
 
-        tis = session.query(TaskInstance).filter(
+        tis = session.query(TaskInstance).options(joinedload(TaskInstance.task_tags)).filter(

Review comment:
       Make this an optional flag on this method -- the scheduler doesn't care about tags so shouldn't pay the cost of loading them.

##########
File path: airflow/models/baseoperator.py
##########
@@ -243,6 +243,8 @@ class derived from this one results in the creation of a task object,
     :param do_xcom_push: if True, an XCom is pushed containing the Operator's
         result
     :type do_xcom_push: bool
+    :param task_tags: List of tags used to identify this task

Review comment:
       ```suggestion
       :param tags: List of tags used to identify this task
   ```
   
   We don't need to say "task" here -- it's already on a task.

##########
File path: airflow/models/baseoperator.py
##########
@@ -359,6 +363,8 @@ def __init__(
         self.email = email
         self.email_on_retry = email_on_retry
         self.email_on_failure = email_on_failure
+        self._task_tags: Optional[List[str]] = []
+        self.task_tags = task_tags or []

Review comment:
       
   ```suggestion
           self.task_tags = task_tags
   ```
   
   The setter already handles None.

##########
File path: airflow/migrations/versions/7c1bef9cb7b9_merge_heads.py
##########
@@ -0,0 +1,41 @@
+#
+# 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.
+
+"""Merge heads
+
+Revision ID: 7c1bef9cb7b9
+Revises: 3c20cacc0044, 19d8a998c007
+Create Date: 2020-06-08 14:09:06.676386
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '7c1bef9cb7b9'
+down_revision = ('3c20cacc0044', '19d8a998c007')

Review comment:
       There should not be a merge migration. Please rebase your change and update the `down_revision` on your new migratin instead.

##########
File path: airflow/models/taskinstance.py
##########
@@ -501,6 +524,13 @@ def refresh_from_task(self, task, pool_override=None):
         self.max_tries = task.retries
         self.executor_config = task.executor_config
         self.operator = task.__class__.__name__
+        # Mocks tags attribute is not iterable
+        if isinstance(task.task_tags, coll.Iterable):
+            for tag in task.task_tags:
+                self.task_tags.append(
+                    TaskTag(name=tag, dag_id=self.dag_id,
+                            task_id=self.task_id, execution_date=self.execution_date)
+                )

Review comment:
       Doesn't SQLA handle this automatically for the relationship?
   ```suggestion
                   self.task_tags.append(TaskTag(name=tag))
   ```

##########
File path: airflow/models/taskinstance.py
##########
@@ -501,6 +524,13 @@ def refresh_from_task(self, task, pool_override=None):
         self.max_tries = task.retries
         self.executor_config = task.executor_config
         self.operator = task.__class__.__name__
+        # Mocks tags attribute is not iterable

Review comment:
       Don't put support code for tests in the real class -- provide a better mock instead.




----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org