hunk.code_window:         # is the execution date in the future?
           if self.execution_date > datetime.now():
               return False
           # is the task still in the retry waiting period?
           elif self.state == State.UP_FOR_RETRY and not self.ready_for_retry():
               return False
           # does the task have an end_date prior to the execution date?
           elif self.task.end_date and self.execution_date > self.task.end_date:
  
hunk.labels: keep keep keep keep replace keep keep keep
hunk.after_edit:         elif self.is_premature():
  
hunk.file_path: airflow/models.py
hunk.type: replace
file: from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  from __future__ import unicode_literals
  
  from builtins import object
  import imp
  import inspect
  import logging
  import os
  import re
  import sys
  from itertools import chain
  merge = chain.from_iterable
  
  from airflow import configuration
  
  class AirflowPluginException(Exception):
      pass
  
  class AirflowPlugin(object):
      name = None
      operators = []
      hooks = []
      executors = []
      macros = []
      admin_views = []
      flask_blueprints = []
      menu_links = []
  
      @classmethod
      def validate(cls):
          if not cls.name:
              raise AirflowPluginException("Your plugin needs a name.")
  
  
  plugins_folder = configuration.get('core', 'plugins_folder')
  if not plugins_folder:
      plugins_folder = configuration.get('core', 'airflow_home') + '/plugins'
  plugins_folder = os.path.expanduser(plugins_folder)
  
  if plugins_folder not in sys.path:
      sys.path.append(plugins_folder)
  
  plugins = []
  
  norm_pattern = re.compile(r'[/|.]')
  
  # Crawl through the plugins folder to find AirflowPlugin derivatives
  for root, dirs, files in os.walk(plugins_folder, followlinks=True):
      for f in files:
          try:
              filepath = os.path.join(root, f)
              if not os.path.isfile(filepath):
                  continue
              mod_name, file_ext = os.path.splitext(
                  os.path.split(filepath)[-1])
              if file_ext != '.py':
                  continue
  
              # normalize root path as namespace
              namespace = '_'.join([re.sub(norm_pattern, '__', root), mod_name])
  
              m = imp.load_source(namespace, filepath)
              for obj in list(m.__dict__.values()):
                  if (
                          inspect.isclass(obj) and
                          issubclass(obj, AirflowPlugin) and
                          obj is not AirflowPlugin):
                      obj.validate()
                      if obj not in plugins:
                          plugins.append(obj)
  
          except Exception as e:
              logging.exception(e)
              logging.error('Failed to import plugin ' + filepath)
  
  operators = merge([p.operators for p in plugins])
  hooks = merge([p.hooks for p in plugins])
  executors = merge([p.executors for p in plugins])
  macros = merge([p.macros for p in plugins])
  admin_views = merge([p.admin_views for p in plugins])
  flask_blueprints = merge([p.flask_blueprints for p in plugins])
  menu_links = merge([p.menu_links for p in plugins])
  