Find the latest file in SFTP directory in Apache Airflow sensor

Viewed 1568

I have two conditions need to fulfill for poking:

  1. Check if there are files landed in specific directory
  2. If there are files, only check on latest files landed

If there are latest files, then, poking is successful to move to next task or not.

My code :

class SFTPSensor(BaseSensorOperator):

    template_fields = ('path',)

    @apply_defaults
    def __init__(self, path, sftp_conn_id='sftp_default', *args, **kwargs): # pylint: disable=keyword-arg-before-vararg
        super(SFTPSensor, self).__init__(*args, **kwargs) # pylint: disable=super-with-arguments
        self.path = path
        self.hook = None
        self.sftp_conn_id = sftp_conn_id

    def poke(self, context): # pylint: disable=unused-argument
        self.hook = SFTPHook(self.sftp_conn_id)
        self.log.info('Poking for %s', self.path)
        try:
            files_list = self.hook.list_directory(self.path)

            print("List of files: ",files_list)
            print("LENGTH OF FILES {}".format(len(files_list)))

            if len(files_list) > 0: # Condition 1 is fulfilled
                return True

        except IOError as e: # pylint: disable=invalid-name
            if e.errno != SFTP_NO_SUCH_FILE:
                raise e
            return False
        self.hook.close_conn()
        return True

From the above code, condition 1 is successful. I'm not quite sure how to do condition 2. Kindly help!

1 Answers

Use describe_directory to get a file list including timestamps:

files_list = self.hook.describe_directory(self.path)

Then just iterate the list and find the file with the highest 'modify' element.

Related