| 178 | |
| 179 | @contextmanager |
| 180 | def patch_fileio(global_vars=None): |
| 181 | if getattr(patch_fileio, '_patched', False): |
| 182 | # Only patch once, avoid error caused by patch nestly. |
| 183 | yield |
| 184 | return |
| 185 | import builtins |
| 186 | |
| 187 | @patch_func(builtins, 'open') |
| 188 | def open(file, mode='r', *args, **kwargs): |
| 189 | backend = get_file_backend(file) |
| 190 | if isinstance(backend, LocalBackend): |
| 191 | return open._fallback(file, mode, *args, **kwargs) |
| 192 | if 'b' in mode: |
| 193 | return io.BytesIO(backend.get(file, *args, **kwargs)) |
| 194 | else: |
| 195 | return io.StringIO(backend.get_text(file, *args, **kwargs)) |
| 196 | |
| 197 | if global_vars is not None and 'open' in global_vars: |
| 198 | bak_open = global_vars['open'] |
| 199 | global_vars['open'] = builtins.open |
| 200 | |
| 201 | import os |
| 202 | |
| 203 | @patch_func(os.path, 'join') |
| 204 | def join(a, *paths): |
| 205 | backend = get_file_backend(a) |
| 206 | if isinstance(backend, LocalBackend): |
| 207 | return join._fallback(a, *paths) |
| 208 | paths = [item for item in paths if len(item) > 0] |
| 209 | return backend.join_path(a, *paths) |
| 210 | |
| 211 | @patch_func(os.path, 'isdir') |
| 212 | def isdir(path): |
| 213 | backend = get_file_backend(path) |
| 214 | if isinstance(backend, LocalBackend): |
| 215 | return isdir._fallback(path) |
| 216 | return backend.isdir(path) |
| 217 | |
| 218 | @patch_func(os.path, 'isfile') |
| 219 | def isfile(path): |
| 220 | backend = get_file_backend(path) |
| 221 | if isinstance(backend, LocalBackend): |
| 222 | return isfile._fallback(path) |
| 223 | return backend.isfile(path) |
| 224 | |
| 225 | @patch_func(os.path, 'exists') |
| 226 | def exists(path): |
| 227 | backend = get_file_backend(path) |
| 228 | if isinstance(backend, LocalBackend): |
| 229 | return exists._fallback(path) |
| 230 | return backend.exists(path) |
| 231 | |
| 232 | @patch_func(os, 'listdir') |
| 233 | def listdir(path): |
| 234 | backend = get_file_backend(path) |
| 235 | if isinstance(backend, LocalBackend): |
| 236 | return listdir._fallback(path) |
| 237 | return backend.list_dir_or_file(path) |