hunk.code_window:             elif ti.is_runnable(flag_upstream_failed=True):
                   self.logger.debug('Queuing task: {}'.format(ti))
                   queue.put((ti.key, pickle_id))
               else:
  
hunk.labels: keep keep add keep
hunk.after_edit:             elif ti.is_premature():
                   continue
  
hunk.file_path: airflow/jobs.py
hunk.type: add
file: import logging
  
  from airflow.hooks import SqliteHook
  from airflow.models import BaseOperator
  from airflow.utils.decorators import apply_defaults
  
  
  class SqliteOperator(BaseOperator):
      """
      Executes sql code in a specific Sqlite database
  
      :param sqlite_conn_id: reference to a specific sqlite database
      :type sqlite_conn_id: string
      :param sql: the sql code to be executed
      :type sql: string or string pointing to a template file. File must have
          a '.sql' extensions.
      """
  
      template_fields = ('sql',)
      template_ext = ('.sql',)
      ui_color = '#cdaaed'
  
      @apply_defaults
      def __init__(
              self, sql, sqlite_conn_id='sqlite_default', parameters=None,
              *args, **kwargs):
          super(SqliteOperator, self).__init__(*args, **kwargs)
          self.sqlite_conn_id = sqlite_conn_id
          self.sql = sql
          self.parameters = parameters or []
  
      def execute(self, context):
          logging.info('Executing: ' + self.sql)
          hook = SqliteHook(sqlite_conn_id=self.sqlite_conn_id)
          hook.run(self.sql, parameters=self.parameters)
  