We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
File creation fails if no executor has been loaded
```python
In [53]: file = File('first.txt')
---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
<ipython-input-53-b5dc63ca042a> in <module>()
----> 1 file = File('first.txt')

~/ci/parsl/parsl/data_provider/files.py in __init__(self, url, dman, cache, caching_dir)
     40         self.path = parsed_url.path
     41         self.filename = os.path.basename(self.path)
---> 42         self.dman = dman if dman else DataManager.get_data_manager()
     43         self.data_future = {}
     44         if self.scheme == 'globus':

~/ci/parsl/parsl/data_provider/data_manager.py in get_data_manager(cls, max_threads, executors)
     24     def get_data_manager(cls, max_threads=None, executors=None):
     25         if cls.default_data_manager is None:
---> 26             cls.default_data_manager = DataManager(max_threads=max_threads, executors=executors)
     27         return cls.default_data_manager
     28

~/ci/parsl/parsl/data_provider/data_manager.py in __init__(self, max_threads, executors)
     37
     38         self.label = 'data_manager'
---> 39         self.executors = {e.label: e for e in executors}
     40         self.max_threads = max_threads
     41         self.files = []

TypeError: 'NoneType' object is not iterable
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `parsl/data_provider/data_manager.py`
Content:
```
1 import os
2 import logging
3 import requests
4 import ftplib
5 import concurrent.futures as cf
6 from parsl.data_provider.scheme import GlobusScheme
7 from parsl.executors.base import ParslExecutor
8 from parsl.data_provider.globus import get_globus
9 from parsl.app.app import App
10 
11 logger = logging.getLogger(__name__)
12 
13 
14 class DataManager(ParslExecutor):
15     """The DataManager is responsible for transferring input and output data.
16 
17     It uses the Executor interface, where staging tasks are submitted
18     to it, and DataFutures are returned.
19     """
20 
21     default_data_manager = None
22 
23     @classmethod
24     def get_data_manager(cls, max_threads=None, executors=None):
25         if cls.default_data_manager is None:
26             cls.default_data_manager = DataManager(max_threads=max_threads, executors=executors)
27         return cls.default_data_manager
28 
29     def __init__(self, max_threads=10, executors=None):
30         """Initialize the DataManager.
31 
32         Kwargs:
33            - max_threads (int): Number of threads. Default is 10.
34            - executors (list of Executors): Executors for which data transfer will be managed.
35         """
36         self._scaling_enabled = False
37 
38         self.label = 'data_manager'
39         self.executors = {e.label: e for e in executors}
40         self.max_threads = max_threads
41         self.files = []
42         self.globus = None
43         self.managed = True
44 
45     def start(self):
46         self.executor = cf.ThreadPoolExecutor(max_workers=self.max_threads)
47 
48     def submit(self, *args, **kwargs):
49         """Submit a staging app. All optimization should be here."""
50         return self.executor.submit(*args, **kwargs)
51 
52     def scale_in(self, blocks, *args, **kwargs):
53         pass
54 
55     def scale_out(self, *args, **kwargs):
56         pass
57 
58     def shutdown(self, block=False):
59         """Shutdown the ThreadPool.
60 
61         Kwargs:
62             - block (bool): To block for confirmations or not
63 
64         """
65         x = self.executor.shutdown(wait=block)
66         logger.debug("Done with executor shutdown")
67         return x
68 
69     @property
70     def scaling_enabled(self):
71         return self._scaling_enabled
72 
73     def add_file(self, file):
74         if file.scheme == 'globus':
75             if not self.globus:
76                 self.globus = get_globus()
77             # keep a list of all remote files for optimization purposes (TODO)
78             self.files.append(file)
79             self._set_local_path(file)
80 
81     def _set_local_path(self, file):
82         globus_ep = self._get_globus_endpoint()
83         file.local_path = os.path.join(globus_ep['working_dir'], file.filename)
84 
85     def _get_globus_endpoint(self, executor_label=None):
86         for executor in self.executors.values():
87             if executor_label is None or executor.label == executor_label:
88                 for scheme in executor.storage_access:
89                     if isinstance(scheme, GlobusScheme):
90                         working_dir = os.path.normpath(executor.working_dir)
91                         if scheme.endpoint_path and scheme.local_path:
92                             endpoint_path = os.path.normpath(scheme.endpoint_path)
93                             local_path = os.path.normpath(scheme.local_path)
94                             common_path = os.path.commonpath((local_path, working_dir))
95                             if local_path != common_path:
96                                 raise Exception('"local_path" must be equal or an absolute subpath of "working_dir"')
97                             relative_path = os.path.relpath(working_dir, common_path)
98                             endpoint_path = os.path.join(endpoint_path, relative_path)
99                         else:
100                             endpoint_path = working_dir
101                         return {'endpoint_uuid': scheme.endpoint_uuid,
102                                 'endpoint_path': endpoint_path,
103                                 'working_dir': working_dir}
104         raise Exception('No executor with a Globus endpoint and working_dir defined')
105 
106     def stage_in(self, file, executor):
107         """Transport the file from the input source to the executor.
108 
109         This function returns a DataFuture.
110 
111         Args:
112             - self
113             - file (File) : file to stage in
114             - executor (str) : an executor the file is going to be staged in to.
115                                 If the executor argument is not specified for a file
116                                 with 'globus' scheme, the file will be staged in to
117                                 the first executor with the "globus" key in a config.
118         """
119 
120         if file.scheme == 'file':
121             stage_in_app = self._file_stage_in_app()
122             app_fut = stage_in_app(outputs=[file])
123             return app_fut._outputs[0]
124         elif file.scheme == 'ftp':
125             working_dir = self.executors[executor].working_dir
126             stage_in_app = self._ftp_stage_in_app(executor=executor)
127             app_fut = stage_in_app(working_dir, outputs=[file])
128             return app_fut._outputs[0]
129         elif file.scheme == 'http' or file.scheme == 'https':
130             working_dir = self.executors[executor].working_dir
131             stage_in_app = self._http_stage_in_app(executor=executor)
132             app_fut = stage_in_app(working_dir, outputs=[file])
133             return app_fut._outputs[0]
134         elif file.scheme == 'globus':
135             globus_ep = self._get_globus_endpoint(executor)
136             stage_in_app = self._globus_stage_in_app()
137             app_fut = stage_in_app(globus_ep, outputs=[file])
138             return app_fut._outputs[0]
139         else:
140             raise Exception('Staging in with unknown file scheme {} is not supported'.format(file.scheme))
141 
142     def _file_stage_in_app(self):
143         return App("python", executors=['data_manager'])(self._file_stage_in)
144 
145     def _file_stage_in(self, outputs=[]):
146         pass
147 
148     def _ftp_stage_in_app(self, executor):
149         return App("python", executors=[executor])(self._ftp_stage_in)
150 
151     def _ftp_stage_in(self, working_dir, outputs=[]):
152         file = outputs[0]
153         if working_dir:
154             os.makedirs(working_dir, exist_ok=True)
155             file.local_path = os.path.join(working_dir, file.filename)
156         else:
157             file.local_path = file.filename
158         with open(file.local_path, 'wb') as f:
159             ftp = ftplib.FTP(file.netloc)
160             ftp.login()
161             ftp.cwd(os.path.dirname(file.path))
162             ftp.retrbinary('RETR {}'.format(file.filename), f.write)
163             ftp.quit()
164 
165     def _http_stage_in_app(self, executor):
166         return App("python", executors=[executor])(self._http_stage_in)
167 
168     def _http_stage_in(self, working_dir, outputs=[]):
169         file = outputs[0]
170         if working_dir:
171             os.makedirs(working_dir, exist_ok=True)
172             file.local_path = os.path.join(working_dir, file.filename)
173         else:
174             file.local_path = file.filename
175         resp = requests.get(file.url, stream=True)
176         with open(file.local_path, 'wb') as f:
177             for chunk in resp.iter_content(chunk_size=1024):
178                 if chunk:
179                     f.write(chunk)
180 
181     def _globus_stage_in_app(self):
182         return App("python", executors=['data_manager'])(self._globus_stage_in)
183 
184     def _globus_stage_in(self, globus_ep, outputs=[]):
185         file = outputs[0]
186         file.local_path = os.path.join(
187                 globus_ep['working_dir'], file.filename)
188         dst_path = os.path.join(
189                 globus_ep['endpoint_path'], file.filename)
190         self.globus.transfer_file(
191                 file.netloc, globus_ep['endpoint_uuid'],
192                 file.path, dst_path)
193 
194     def stage_out(self, file, executor):
195         """Transport the file from the local filesystem to the remote Globus endpoint.
196 
197         This function returns a DataFuture.
198 
199         Args:
200             - self
201             - file (File) - file to stage out
202             - executor (str) - Which executor the file is going to be staged out from.
203                                 If the executor argument is not specified for a file
204                                 with the 'globus' scheme, the file will be staged in to
205                                 the first executor with the "globus" key in a config.
206         """
207 
208         if file.scheme == 'file':
209             stage_out_app = self._file_stage_out_app()
210             return stage_out_app()
211         elif file.scheme == 'http' or file.scheme == 'https':
212             raise Exception('HTTP/HTTPS file staging out is not supported')
213         elif file.scheme == 'ftp':
214             raise Exception('FTP file staging out is not supported')
215         elif file.scheme == 'globus':
216             globus_ep = self._get_globus_endpoint(executor)
217             stage_out_app = self._globus_stage_out_app()
218             return stage_out_app(globus_ep, inputs=[file])
219         else:
220             raise Exception('Staging out with unknown file scheme {} is not supported'.format(file.scheme))
221 
222     def _file_stage_out_app(self):
223         return App("python", executors=['data_manager'])(self._file_stage_out)
224 
225     def _file_stage_out(self):
226         pass
227 
228     def _globus_stage_out_app(self):
229         return App("python", executors=['data_manager'])(self._globus_stage_out)
230 
231     def _globus_stage_out(self, globus_ep, inputs=[]):
232         file = inputs[0]
233         src_path = os.path.join(globus_ep['endpoint_path'], file.filename)
234         self.globus.transfer_file(
235             globus_ep['endpoint_uuid'], file.netloc,
236             src_path, file.path
237         )
238
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/parsl/data_provider/data_manager.py b/parsl/data_provider/data_manager.py
--- a/parsl/data_provider/data_manager.py
+++ b/parsl/data_provider/data_manager.py
@@ -36,6 +36,8 @@
         self._scaling_enabled = False
 
         self.label = 'data_manager'
+        if executors is None:
+            executors = []
         self.executors = {e.label: e for e in executors}
         self.max_threads = max_threads
         self.files = []

{"golden_diff": "diff --git a/parsl/data_provider/data_manager.py b/parsl/data_provider/data_manager.py\n--- a/parsl/data_provider/data_manager.py\n+++ b/parsl/data_provider/data_manager.py\n@@ -36,6 +36,8 @@\n         self._scaling_enabled = False\n \n         self.label = 'data_manager'\n+        if executors is None:\n+            executors = []\n         self.executors = {e.label: e for e in executors}\n         self.max_threads = max_threads\n         self.files = []\n", "issue": "File creation fails if no executor has been loaded\n```python\r\nIn [53]: file = File('first.txt')\r\n---------------------------------------------------------------------------\r\nTypeError                                 Traceback (most recent call last)\r\n<ipython-input-53-b5dc63ca042a> in <module>()\r\n----> 1 file = File('first.txt')\r\n\r\n~/ci/parsl/parsl/data_provider/files.py in __init__(self, url, dman, cache, caching_dir)\r\n     40         self.path = parsed_url.path\r\n     41         self.filename = os.path.basename(self.path)\r\n---> 42         self.dman = dman if dman else DataManager.get_data_manager()\r\n     43         self.data_future = {}\r\n     44         if self.scheme == 'globus':\r\n\r\n~/ci/parsl/parsl/data_provider/data_manager.py in get_data_manager(cls, max_threads, executors)\r\n     24     def get_data_manager(cls, max_threads=None, executors=None):\r\n     25         if cls.default_data_manager is None:\r\n---> 26             cls.default_data_manager = DataManager(max_threads=max_threads, executors=executors)\r\n     27         return cls.default_data_manager\r\n     28\r\n\r\n~/ci/parsl/parsl/data_provider/data_manager.py in __init__(self, max_threads, executors)\r\n     37\r\n     38         self.label = 'data_manager'\r\n---> 39         self.executors = {e.label: e for e in executors}\r\n     40         self.max_threads = max_threads\r\n     41         self.files = []\r\n\r\nTypeError: 'NoneType' object is not iterable\r\n```\n", "before_files": [{"content": "import os\nimport logging\nimport requests\nimport ftplib\nimport concurrent.futures as cf\nfrom parsl.data_provider.scheme import GlobusScheme\nfrom parsl.executors.base import ParslExecutor\nfrom parsl.data_provider.globus import get_globus\nfrom parsl.app.app import App\n\nlogger = logging.getLogger(__name__)\n\n\nclass DataManager(ParslExecutor):\n    \"\"\"The DataManager is responsible for transferring input and output data.\n\n    It uses the Executor interface, where staging tasks are submitted\n    to it, and DataFutures are returned.\n    \"\"\"\n\n    default_data_manager = None\n\n    @classmethod\n    def get_data_manager(cls, max_threads=None, executors=None):\n        if cls.default_data_manager is None:\n            cls.default_data_manager = DataManager(max_threads=max_threads, executors=executors)\n        return cls.default_data_manager\n\n    def __init__(self, max_threads=10, executors=None):\n        \"\"\"Initialize the DataManager.\n\n        Kwargs:\n           - max_threads (int): Number of threads. Default is 10.\n           - executors (list of Executors): Executors for which data transfer will be managed.\n        \"\"\"\n        self._scaling_enabled = False\n\n        self.label = 'data_manager'\n        self.executors = {e.label: e for e in executors}\n        self.max_threads = max_threads\n        self.files = []\n        self.globus = None\n        self.managed = True\n\n    def start(self):\n        self.executor = cf.ThreadPoolExecutor(max_workers=self.max_threads)\n\n    def submit(self, *args, **kwargs):\n        \"\"\"Submit a staging app. All optimization should be here.\"\"\"\n        return self.executor.submit(*args, **kwargs)\n\n    def scale_in(self, blocks, *args, **kwargs):\n        pass\n\n    def scale_out(self, *args, **kwargs):\n        pass\n\n    def shutdown(self, block=False):\n        \"\"\"Shutdown the ThreadPool.\n\n        Kwargs:\n            - block (bool): To block for confirmations or not\n\n        \"\"\"\n        x = self.executor.shutdown(wait=block)\n        logger.debug(\"Done with executor shutdown\")\n        return x\n\n    @property\n    def scaling_enabled(self):\n        return self._scaling_enabled\n\n    def add_file(self, file):\n        if file.scheme == 'globus':\n            if not self.globus:\n                self.globus = get_globus()\n            # keep a list of all remote files for optimization purposes (TODO)\n            self.files.append(file)\n            self._set_local_path(file)\n\n    def _set_local_path(self, file):\n        globus_ep = self._get_globus_endpoint()\n        file.local_path = os.path.join(globus_ep['working_dir'], file.filename)\n\n    def _get_globus_endpoint(self, executor_label=None):\n        for executor in self.executors.values():\n            if executor_label is None or executor.label == executor_label:\n                for scheme in executor.storage_access:\n                    if isinstance(scheme, GlobusScheme):\n                        working_dir = os.path.normpath(executor.working_dir)\n                        if scheme.endpoint_path and scheme.local_path:\n                            endpoint_path = os.path.normpath(scheme.endpoint_path)\n                            local_path = os.path.normpath(scheme.local_path)\n                            common_path = os.path.commonpath((local_path, working_dir))\n                            if local_path != common_path:\n                                raise Exception('\"local_path\" must be equal or an absolute subpath of \"working_dir\"')\n                            relative_path = os.path.relpath(working_dir, common_path)\n                            endpoint_path = os.path.join(endpoint_path, relative_path)\n                        else:\n                            endpoint_path = working_dir\n                        return {'endpoint_uuid': scheme.endpoint_uuid,\n                                'endpoint_path': endpoint_path,\n                                'working_dir': working_dir}\n        raise Exception('No executor with a Globus endpoint and working_dir defined')\n\n    def stage_in(self, file, executor):\n        \"\"\"Transport the file from the input source to the executor.\n\n        This function returns a DataFuture.\n\n        Args:\n            - self\n            - file (File) : file to stage in\n            - executor (str) : an executor the file is going to be staged in to.\n                                If the executor argument is not specified for a file\n                                with 'globus' scheme, the file will be staged in to\n                                the first executor with the \"globus\" key in a config.\n        \"\"\"\n\n        if file.scheme == 'file':\n            stage_in_app = self._file_stage_in_app()\n            app_fut = stage_in_app(outputs=[file])\n            return app_fut._outputs[0]\n        elif file.scheme == 'ftp':\n            working_dir = self.executors[executor].working_dir\n            stage_in_app = self._ftp_stage_in_app(executor=executor)\n            app_fut = stage_in_app(working_dir, outputs=[file])\n            return app_fut._outputs[0]\n        elif file.scheme == 'http' or file.scheme == 'https':\n            working_dir = self.executors[executor].working_dir\n            stage_in_app = self._http_stage_in_app(executor=executor)\n            app_fut = stage_in_app(working_dir, outputs=[file])\n            return app_fut._outputs[0]\n        elif file.scheme == 'globus':\n            globus_ep = self._get_globus_endpoint(executor)\n            stage_in_app = self._globus_stage_in_app()\n            app_fut = stage_in_app(globus_ep, outputs=[file])\n            return app_fut._outputs[0]\n        else:\n            raise Exception('Staging in with unknown file scheme {} is not supported'.format(file.scheme))\n\n    def _file_stage_in_app(self):\n        return App(\"python\", executors=['data_manager'])(self._file_stage_in)\n\n    def _file_stage_in(self, outputs=[]):\n        pass\n\n    def _ftp_stage_in_app(self, executor):\n        return App(\"python\", executors=[executor])(self._ftp_stage_in)\n\n    def _ftp_stage_in(self, working_dir, outputs=[]):\n        file = outputs[0]\n        if working_dir:\n            os.makedirs(working_dir, exist_ok=True)\n            file.local_path = os.path.join(working_dir, file.filename)\n        else:\n            file.local_path = file.filename\n        with open(file.local_path, 'wb') as f:\n            ftp = ftplib.FTP(file.netloc)\n            ftp.login()\n            ftp.cwd(os.path.dirname(file.path))\n            ftp.retrbinary('RETR {}'.format(file.filename), f.write)\n            ftp.quit()\n\n    def _http_stage_in_app(self, executor):\n        return App(\"python\", executors=[executor])(self._http_stage_in)\n\n    def _http_stage_in(self, working_dir, outputs=[]):\n        file = outputs[0]\n        if working_dir:\n            os.makedirs(working_dir, exist_ok=True)\n            file.local_path = os.path.join(working_dir, file.filename)\n        else:\n            file.local_path = file.filename\n        resp = requests.get(file.url, stream=True)\n        with open(file.local_path, 'wb') as f:\n            for chunk in resp.iter_content(chunk_size=1024):\n                if chunk:\n                    f.write(chunk)\n\n    def _globus_stage_in_app(self):\n        return App(\"python\", executors=['data_manager'])(self._globus_stage_in)\n\n    def _globus_stage_in(self, globus_ep, outputs=[]):\n        file = outputs[0]\n        file.local_path = os.path.join(\n                globus_ep['working_dir'], file.filename)\n        dst_path = os.path.join(\n                globus_ep['endpoint_path'], file.filename)\n        self.globus.transfer_file(\n                file.netloc, globus_ep['endpoint_uuid'],\n                file.path, dst_path)\n\n    def stage_out(self, file, executor):\n        \"\"\"Transport the file from the local filesystem to the remote Globus endpoint.\n\n        This function returns a DataFuture.\n\n        Args:\n            - self\n            - file (File) - file to stage out\n            - executor (str) - Which executor the file is going to be staged out from.\n                                If the executor argument is not specified for a file\n                                with the 'globus' scheme, the file will be staged in to\n                                the first executor with the \"globus\" key in a config.\n        \"\"\"\n\n        if file.scheme == 'file':\n            stage_out_app = self._file_stage_out_app()\n            return stage_out_app()\n        elif file.scheme == 'http' or file.scheme == 'https':\n            raise Exception('HTTP/HTTPS file staging out is not supported')\n        elif file.scheme == 'ftp':\n            raise Exception('FTP file staging out is not supported')\n        elif file.scheme == 'globus':\n            globus_ep = self._get_globus_endpoint(executor)\n            stage_out_app = self._globus_stage_out_app()\n            return stage_out_app(globus_ep, inputs=[file])\n        else:\n            raise Exception('Staging out with unknown file scheme {} is not supported'.format(file.scheme))\n\n    def _file_stage_out_app(self):\n        return App(\"python\", executors=['data_manager'])(self._file_stage_out)\n\n    def _file_stage_out(self):\n        pass\n\n    def _globus_stage_out_app(self):\n        return App(\"python\", executors=['data_manager'])(self._globus_stage_out)\n\n    def _globus_stage_out(self, globus_ep, inputs=[]):\n        file = inputs[0]\n        src_path = os.path.join(globus_ep['endpoint_path'], file.filename)\n        self.globus.transfer_file(\n            globus_ep['endpoint_uuid'], file.netloc,\n            src_path, file.path\n        )\n", "path": "parsl/data_provider/data_manager.py"}], "after_files": [{"content": "import os\nimport logging\nimport requests\nimport ftplib\nimport concurrent.futures as cf\nfrom parsl.data_provider.scheme import GlobusScheme\nfrom parsl.executors.base import ParslExecutor\nfrom parsl.data_provider.globus import get_globus\nfrom parsl.app.app import App\n\nlogger = logging.getLogger(__name__)\n\n\nclass DataManager(ParslExecutor):\n    \"\"\"The DataManager is responsible for transferring input and output data.\n\n    It uses the Executor interface, where staging tasks are submitted\n    to it, and DataFutures are returned.\n    \"\"\"\n\n    default_data_manager = None\n\n    @classmethod\n    def get_data_manager(cls, max_threads=None, executors=None):\n        if cls.default_data_manager is None:\n            cls.default_data_manager = DataManager(max_threads=max_threads, executors=executors)\n        return cls.default_data_manager\n\n    def __init__(self, max_threads=10, executors=None):\n        \"\"\"Initialize the DataManager.\n\n        Kwargs:\n           - max_threads (int): Number of threads. Default is 10.\n           - executors (list of Executors): Executors for which data transfer will be managed.\n        \"\"\"\n        self._scaling_enabled = False\n\n        self.label = 'data_manager'\n        if executors is None:\n            executors = []\n        self.executors = {e.label: e for e in executors}\n        self.max_threads = max_threads\n        self.files = []\n        self.globus = None\n        self.managed = True\n\n    def start(self):\n        self.executor = cf.ThreadPoolExecutor(max_workers=self.max_threads)\n\n    def submit(self, *args, **kwargs):\n        \"\"\"Submit a staging app. All optimization should be here.\"\"\"\n        return self.executor.submit(*args, **kwargs)\n\n    def scale_in(self, blocks, *args, **kwargs):\n        pass\n\n    def scale_out(self, *args, **kwargs):\n        pass\n\n    def shutdown(self, block=False):\n        \"\"\"Shutdown the ThreadPool.\n\n        Kwargs:\n            - block (bool): To block for confirmations or not\n\n        \"\"\"\n        x = self.executor.shutdown(wait=block)\n        logger.debug(\"Done with executor shutdown\")\n        return x\n\n    @property\n    def scaling_enabled(self):\n        return self._scaling_enabled\n\n    def add_file(self, file):\n        if file.scheme == 'globus':\n            if not self.globus:\n                self.globus = get_globus()\n            # keep a list of all remote files for optimization purposes (TODO)\n            self.files.append(file)\n            self._set_local_path(file)\n\n    def _set_local_path(self, file):\n        globus_ep = self._get_globus_endpoint()\n        file.local_path = os.path.join(globus_ep['working_dir'], file.filename)\n\n    def _get_globus_endpoint(self, executor_label=None):\n        for executor in self.executors.values():\n            if executor_label is None or executor.label == executor_label:\n                for scheme in executor.storage_access:\n                    if isinstance(scheme, GlobusScheme):\n                        working_dir = os.path.normpath(executor.working_dir)\n                        if scheme.endpoint_path and scheme.local_path:\n                            endpoint_path = os.path.normpath(scheme.endpoint_path)\n                            local_path = os.path.normpath(scheme.local_path)\n                            common_path = os.path.commonpath((local_path, working_dir))\n                            if local_path != common_path:\n                                raise Exception('\"local_path\" must be equal or an absolute subpath of \"working_dir\"')\n                            relative_path = os.path.relpath(working_dir, common_path)\n                            endpoint_path = os.path.join(endpoint_path, relative_path)\n                        else:\n                            endpoint_path = working_dir\n                        return {'endpoint_uuid': scheme.endpoint_uuid,\n                                'endpoint_path': endpoint_path,\n                                'working_dir': working_dir}\n        raise Exception('No executor with a Globus endpoint and working_dir defined')\n\n    def stage_in(self, file, executor):\n        \"\"\"Transport the file from the input source to the executor.\n\n        This function returns a DataFuture.\n\n        Args:\n            - self\n            - file (File) : file to stage in\n            - executor (str) : an executor the file is going to be staged in to.\n                                If the executor argument is not specified for a file\n                                with 'globus' scheme, the file will be staged in to\n                                the first executor with the \"globus\" key in a config.\n        \"\"\"\n\n        if file.scheme == 'file':\n            stage_in_app = self._file_stage_in_app()\n            app_fut = stage_in_app(outputs=[file])\n            return app_fut._outputs[0]\n        elif file.scheme == 'ftp':\n            working_dir = self.executors[executor].working_dir\n            stage_in_app = self._ftp_stage_in_app(executor=executor)\n            app_fut = stage_in_app(working_dir, outputs=[file])\n            return app_fut._outputs[0]\n        elif file.scheme == 'http' or file.scheme == 'https':\n            working_dir = self.executors[executor].working_dir\n            stage_in_app = self._http_stage_in_app(executor=executor)\n            app_fut = stage_in_app(working_dir, outputs=[file])\n            return app_fut._outputs[0]\n        elif file.scheme == 'globus':\n            globus_ep = self._get_globus_endpoint(executor)\n            stage_in_app = self._globus_stage_in_app()\n            app_fut = stage_in_app(globus_ep, outputs=[file])\n            return app_fut._outputs[0]\n        else:\n            raise Exception('Staging in with unknown file scheme {} is not supported'.format(file.scheme))\n\n    def _file_stage_in_app(self):\n        return App(\"python\", executors=['data_manager'])(self._file_stage_in)\n\n    def _file_stage_in(self, outputs=[]):\n        pass\n\n    def _ftp_stage_in_app(self, executor):\n        return App(\"python\", executors=[executor])(self._ftp_stage_in)\n\n    def _ftp_stage_in(self, working_dir, outputs=[]):\n        file = outputs[0]\n        if working_dir:\n            os.makedirs(working_dir, exist_ok=True)\n            file.local_path = os.path.join(working_dir, file.filename)\n        else:\n            file.local_path = file.filename\n        with open(file.local_path, 'wb') as f:\n            ftp = ftplib.FTP(file.netloc)\n            ftp.login()\n            ftp.cwd(os.path.dirname(file.path))\n            ftp.retrbinary('RETR {}'.format(file.filename), f.write)\n            ftp.quit()\n\n    def _http_stage_in_app(self, executor):\n        return App(\"python\", executors=[executor])(self._http_stage_in)\n\n    def _http_stage_in(self, working_dir, outputs=[]):\n        file = outputs[0]\n        if working_dir:\n            os.makedirs(working_dir, exist_ok=True)\n            file.local_path = os.path.join(working_dir, file.filename)\n        else:\n            file.local_path = file.filename\n        resp = requests.get(file.url, stream=True)\n        with open(file.local_path, 'wb') as f:\n            for chunk in resp.iter_content(chunk_size=1024):\n                if chunk:\n                    f.write(chunk)\n\n    def _globus_stage_in_app(self):\n        return App(\"python\", executors=['data_manager'])(self._globus_stage_in)\n\n    def _globus_stage_in(self, globus_ep, outputs=[]):\n        file = outputs[0]\n        file.local_path = os.path.join(\n                globus_ep['working_dir'], file.filename)\n        dst_path = os.path.join(\n                globus_ep['endpoint_path'], file.filename)\n        self.globus.transfer_file(\n                file.netloc, globus_ep['endpoint_uuid'],\n                file.path, dst_path)\n\n    def stage_out(self, file, executor):\n        \"\"\"Transport the file from the local filesystem to the remote Globus endpoint.\n\n        This function returns a DataFuture.\n\n        Args:\n            - self\n            - file (File) - file to stage out\n            - executor (str) - Which executor the file is going to be staged out from.\n                                If the executor argument is not specified for a file\n                                with the 'globus' scheme, the file will be staged in to\n                                the first executor with the \"globus\" key in a config.\n        \"\"\"\n\n        if file.scheme == 'file':\n            stage_out_app = self._file_stage_out_app()\n            return stage_out_app()\n        elif file.scheme == 'http' or file.scheme == 'https':\n            raise Exception('HTTP/HTTPS file staging out is not supported')\n        elif file.scheme == 'ftp':\n            raise Exception('FTP file staging out is not supported')\n        elif file.scheme == 'globus':\n            globus_ep = self._get_globus_endpoint(executor)\n            stage_out_app = self._globus_stage_out_app()\n            return stage_out_app(globus_ep, inputs=[file])\n        else:\n            raise Exception('Staging out with unknown file scheme {} is not supported'.format(file.scheme))\n\n    def _file_stage_out_app(self):\n        return App(\"python\", executors=['data_manager'])(self._file_stage_out)\n\n    def _file_stage_out(self):\n        pass\n\n    def _globus_stage_out_app(self):\n        return App(\"python\", executors=['data_manager'])(self._globus_stage_out)\n\n    def _globus_stage_out(self, globus_ep, inputs=[]):\n        file = inputs[0]\n        src_path = os.path.join(globus_ep['endpoint_path'], file.filename)\n        self.globus.transfer_file(\n            globus_ep['endpoint_uuid'], file.netloc,\n            src_path, file.path\n        )\n", "path": "parsl/data_provider/data_manager.py"}]}