This method is used in DagFileProcessor.process_file - it may be easier to migrate all it's internal calls instead of the whole method.
):
                return []
        except Exception as e:
            self.log.exception(e)
            return []
        
        DagContext.autoregistered_dags.clear()
        if filepath.endswith(".py") or not zipfile.is_zipfile(filepath):
            mods = self._load_modules_from_file(filepath, safe_mode)
        else:
            mods = self._load_modules_from_zip(filepath, safe_mode)
        found_dags = self._process_modules(filepath, mods, file_last_changed_on_disk)
        self.file_last_changed[filepath] = file_last_changed_on_disk
        return found_dags
    def _load_modules_from_file(self, filepath, safe_mode):
        from airflow.models.dag import DagContext
        if not might_contain_dag(filepath, safe_mode):
            
            if not self.has_logged:
                self.has_logged = True
                self.log.info("File %s assumed to contain no DAGs. Skipping.", filepath)
            return []
        self.log.debug("Importing %s", filepath)
        org_mod_name, _ = os.path.splitext(os.path.split(filepath)[-1])
        path_hash = hashlib.sha1(filepath.encode("utf-8")).hexdigest()
        mod_name = f"unusual_prefix_{path_hash}_{org_mod_name}"
        if mod_name in sys.modules:
            del sys.modules[mod_name]
        DagContext.current_autoregister_module_name = mod_name
        def parse(mod_name, filepath):