benjamingregory

from airflow.hooks.ftp_hook import FTPHook

def describe_directory(**kwargs):
    path = kwargs.get('path', None)
    ftp_conn_id = kwargs.get('ftp_conn_id', None)
    filename = kwargs.get('templates_dict').get('filename', None)
    ftp_hook = FTPHook(ftp_conn_id=ftp_conn_id)
    client = ftp_hook.get_conn()
    directory = client.describe_directory(path)