| 123456789101112131415161718192021222324252627282930313233343536373839404142434445 |
- from __future__ import annotations
- import typing
- from ._errors import IllegalDestination
- from ._path import combine, frombase, isbase
- from ._tools import copy_file_data
- if typing.TYPE_CHECKING:
- from ._base import FS
- def copy_file(src_fs: FS, src_path: str, dst_fs: FS, dst_path: str):
- if src_fs is dst_fs and src_path == dst_path:
- raise IllegalDestination(f"cannot copy {src_path!r} to itself")
- with src_fs.open(src_path, "rb") as src_file:
- with dst_fs.open(dst_path, "wb") as dst_file:
- copy_file_data(src_file, dst_file)
- def copy_structure(
- src_fs: FS,
- dst_fs: FS,
- src_root: str = "/",
- dst_root: str = "/",
- ):
- if src_fs is dst_fs and isbase(src_root, dst_root):
- raise IllegalDestination(f"cannot copy {src_fs!r} to itself")
- dst_fs.makedirs(dst_root, recreate=True)
- for dir_path in src_fs.walk.dirs(src_root):
- dst_fs.makedir(combine(dst_root, frombase(src_root, dir_path)), recreate=True)
- def copy_dir(src_fs: FS, src_path: str, dst_fs: FS, dst_path: str):
- copy_structure(src_fs, dst_fs, src_path, dst_path)
- for file_path in src_fs.walk.files(src_path):
- copy_path = combine(dst_path, frombase(src_path, file_path))
- copy_file(src_fs, file_path, dst_fs, copy_path)
- def copy_fs(src_fs: FS, dst_fs: FS):
- copy_dir(src_fs, "/", dst_fs, "/")
|
