https://github.com/apache/airflow/blob/main/airflow/dag_processing/manager.py#L773
"""Returns the dag_director as a string."""
        if isinstance(self._dag_directory, Path):
            return str(self._dag_directory.resolve())
        else:
            return str(self._dag_directory)
    def set_file_paths(self, new_file_paths):
        """
        Update this with a new set of paths to DAG definition files.
        :param new_file_paths: list of paths to DAG definition files
        :return: None
        """
        self._file_paths = new_file_paths
        self._file_path_queue = collections.deque(x for x in self._file_path_queue if x in new_file_paths)
        
        filtered_processors = {}
        for file_path, processor in self._processors.items():
            if file_path in new_file_paths:
                filtered_processors[file_path] = processor
            else:
                self.log.warning("Stopping processor for %s", file_path)
                Stats.decr("dag_processing.processes", tags={"file_path": file_path, "action": "stop"})
                processor.terminate()
                self._file_stats.pop(file_path)
        to_remove = set(self._file_stats.keys()) - set(self._file_paths)
        for key in to_remove:
            
            del self._file_stats[key]
        self._processors = filtered_processors
    def wait_until_finished(self):