'Given a path to a python module or zip file, this method imports
the module and look for dag objects within it.'

def process_file(self, filepath, only_if_updated=True, safe_mode=True):

    found_dags = []
    if (not os.path.isfile(filepath)):
        return found_dags
    try:
        file_last_changed_on_disk = datetime.fromtimestamp(os.path.getmtime(filepath))
        if (only_if_updated and (filepath in self.file_last_changed) and (file_last_changed_on_disk == self.file_last_changed[filepath])):
            return found_dags
    except Exception as e:
        logging.exception(e)
        return found_dags
    mods = []
    if (not zipfile.is_zipfile(filepath)):
        if (safe_mode and os.path.isfile(filepath)):
            with open(filepath, u'rb') as f:
                content = f.read()
                if (not all([(s in content) for s in ('DAG', 'airflow')])):
                    self.file_last_changed[filepath] = file_last_changed_on_disk
                    return found_dags
        self.logger.debug(u'Importing    {}'.format(filepath))
        (org_mod_name, _) = os.path.splitext(os.path.split(filepath)[(-1)])
        mod_name = (((u'unusual_prefix_' + hashlib.sha1(filepath.encode(u'utf-8')).hexdigest()) + u'_') + org_mod_name)
        if (mod_name in sys.modules):
            del sys.modules[mod_name]
        with timeout(configuration.getint(u'core', u'DAGBAG_IMPORT_TIMEOUT')):
            try:
                m = imp.load_source(mod_name, filepath)
                mods.append(m)
            except Exception as e:
                self.logger.exception((u'Failed    to    import:    ' + filepath))
                self.import_errors[filepath] = str(e)
                self.file_last_changed[filepath] = file_last_changed_on_disk
    else:
        zip_file = zipfile.ZipFile(filepath)
        for mod in zip_file.infolist():
            (head, _) = os.path.split(mod.filename)
            (mod_name, ext) = os.path.splitext(mod.filename)
            if ((not head) and ((ext == u'.py') or (ext == u'.pyc'))):
                if (mod_name == u'__init__'):
                    self.logger.warning(u'Found    __init__.{0}    at    root    of    {1}'.format(ext, filepath))
                if safe_mode:
                    with zip_file.open(mod.filename) as zf:
                        self.logger.debug(u'Reading    {}    from    {}'.format(mod.filename, filepath))
                        content = zf.read()
                        if (not all([(s in content) for s in ('DAG', 'airflow')])):
                            self.file_last_changed[filepath] = file_last_changed_on_disk
                            return found_dags
                if (mod_name in sys.modules):
                    del sys.modules[mod_name]
                try:
                    sys.path.insert(0, filepath)
                    m = importlib.import_module(mod_name)
                    mods.append(m)
                except Exception as e:
                    self.logger.exception((u'Failed    to    import:    ' + filepath))
                    self.import_errors[filepath] = str(e)
                    self.file_last_changed[filepath] = file_last_changed_on_disk
    for m in mods:
        for dag in list(m.__dict__.values()):
            if isinstance(dag, DAG):
                if (not dag.full_filepath):
                    dag.full_filepath = filepath
                dag.is_subdag = False
                self.bag_dag(dag, parent_dag=dag, root_dag=dag)
                found_dags.append(dag)
                found_dags += dag.subdags
    self.file_last_changed[filepath] = file_last_changed_on_disk
    return found_dags
