I have two conditions need to fulfill for poking:
- Check if there are files landed in specific directory
- 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!