This method is used in DagFileProcessor.process_file - it may be easier to migrate all it's internal calls instead of the whole method.
dagbag_import_timeout = settings.get_dagbag_import_timeout(filepath)
        if not isinstance(dagbag_import_timeout, (int, float)):
            raise TypeError(
                f"Value ({dagbag_import_timeout}) from get_dagbag_import_timeout must be int or float"
            )
        if dagbag_import_timeout <= 0:  
            return parse(mod_name, filepath)
        timeout_msg = (
            f"DagBag import timeout for {filepath} after {dagbag_import_timeout}s.\n"
            "Please take a look at these docs to improve your DAG import time:\n"
            f"* {get_docs_url('best-practices.html#top-level-python-code')}\n"
            f"* {get_docs_url('best-practices.html#reducing-dag-complexity')}"
        )
        with timeout(dagbag_import_timeout, error_message=timeout_msg):
            return parse(mod_name, filepath)
    def _load_modules_from_zip(self, filepath, safe_mode):
        from airflow.models.dag import DagContext
        mods = []
        with zipfile.ZipFile(filepath) as current_zip_file:
            for zip_info in current_zip_file.infolist():
                head, _ = os.path.split(zip_info.filename)
                mod_name, ext = os.path.splitext(zip_info.filename)
                if ext not in [".py", ".pyc"]:
                    continue
                if head:
                    continue
                if mod_name == "__init__":
                    self.log.warning("Found __init__.%s at root of %s", ext, filepath)
                self.log.debug("Reading %s from %s", zip_info.filename, filepath)
                if not might_contain_dag(zip_info.filename, safe_mode, current_zip_file):