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

        # todo: raise exception?
        if not os.path.isfile(filepath):
            return found_dags

        try:
            # This failed before in what may have been a git sync
            # race condition
            dttm = datetime.fromtimestamp(os.path.getmtime(filepath))
            if only_if_updated \
                    and filepath in self.file_last_changed \
                    and dttm == 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, 'rb') as f:
                    content = f.read()
                    if not all([s in content for s in (b'DAG', b'airflow')]):
                        return found_dags

            self.logger.debug("Importing {}".format(filepath))
            org_mod_name, file_ext = os.path.splitext(os.path.split(filepath)[-1])
            mod_name = 'unusual_prefix_' + org_mod_name

            if mod_name in sys.modules:
                del sys.modules[mod_name]

            with timeout(configuration.getint('core', "DAGBAG_IMPORT_TIMEOUT")):
                try:
                    m = imp.load_source(mod_name, filepath)
                    mods.append(m)
                except Exception as e:
                    self.logger.exception("Failed to import: " + filepath)
                    self.import_errors[filepath] = str(e)
                    self.file_last_changed[filepath] = dttm

        else:
            zip_file = zipfile.ZipFile(filepath)
            for mod in zip_file.infolist():
                head, tail = os.path.split(mod.filename)
                mod_name, ext = os.path.splitext(mod.filename)
                if not head and (ext == '.py' or ext == '.pyc'):
                    if mod_name == '__init__':
                        self.logger.warning("Found __init__.{0} at root of {1}".
                                            format(ext, filepath))

                    if safe_mode:
                        with zip_file.open(mod.filename) as zf:
                            self.logger.debug("Reading {} from {}".
                                              format(mod.filename, filepath))
                            content = zf.read()
                            if not all([s in content for s in (b'DAG', b'airflow')]):
                                # todo: create ignore list
                                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("Failed to import: " + filepath)
                        self.import_errors[filepath] = str(e)
                        self.file_last_changed[filepath] = dttm

        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
                    dag.module_name = m.__name__
                    self.bag_dag(dag, parent_dag=dag, root_dag=dag)
                    found_dags.append(dag)
                    found_dags += dag.subdags

        self.file_last_changed[filepath] = dttm
        return found_dags
Given a path to a python module or zip file, this method imports
dataset/ETHPy150Open airbnb/airflow/airflow/models.py/DagBag.process_file