Skip to content

Commit

Permalink
feat(io): add function to copy a list of files to a target directory
Browse files Browse the repository at this point in the history
Signed-off-by: Cameron Smith <cameron.ray.smith@gmail.com>
  • Loading branch information
cameronraysmith committed Aug 15, 2024
1 parent 86cc9d7 commit eb0b9f9
Showing 1 changed file with 70 additions and 2 deletions.
72 changes: 70 additions & 2 deletions src/pyrovelocity/io/archive.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,13 +3,17 @@
from pathlib import Path

from beartype import beartype
from fsspec import AbstractFileSystem
from beartype.typing import List, Union
from fsspec import AbstractFileSystem, filesystem
from fsspec.implementations.local import LocalFileSystem
from returns.result import Failure, Result, Success

from pyrovelocity.logging import configure_logging

__all__ = ["create_tarball_from_filtered_dir"]
__all__ = [
"create_tarball_from_filtered_dir",
"copy_files_to_directory",
]

logger = configure_logging(__name__)

Expand Down Expand Up @@ -77,3 +81,67 @@ def create_tarball_from_filtered_dir(
f"Failed to create gzipped tarball from {src_dir} due to exception: {e}"
)
return Failure(e)


@beartype
def copy_files_to_directory(
files_to_copy: List[str | Path],
target_directory: str | Path,
) -> Result[None, Exception]:
"""
Copy a list of files to a target directory using fsspec.
Args:
files_to_copy: A list of file paths (str or Path) to copy.
target_directory: The target directory path (str or Path) to copy files into.
Returns:
Result[None, Exception]: A result object encapsulating success or failure.
Examples:
>>> from pathlib import Path
>>> tmp = getfixture('tmp_path') # Get a temporary directory path
>>> source_dir = tmp / 'source'
>>> source_dir.mkdir() # Create a source directory
>>> (source_dir / 'file1.txt').write_text('Content 1') # Create a text file
>>> (source_dir / 'file2.csv').write_text('Content 2') # Create a CSV file
>>> target_dir = tmp / 'target'
>>> files_to_copy = [source_dir / 'file1.txt', source_dir / 'file2.csv']
>>> result = copy_files_to_directory(files_to_copy, target_dir) # Copy files
>>> print(f"Result: {result}") # Check the result
Result: <Success: None>
>>> assert (target_dir / 'file1.txt').exists() # Check if file1.txt was copied
>>> assert (target_dir / 'file2.csv').exists() # Check if file2.csv was copied
>>> with open(target_dir / 'file1.txt', 'r') as f:
... content = f.read()
... assert content == 'Content 1' # Check if content was correctly copied
>>> with open(target_dir / 'file2.csv', 'r') as f:
... content = f.read()
... assert content == 'Content 2' # Check if content was correctly copied
"""
fs: LocalFileSystem = filesystem("file")
target_directory = str(target_directory)

try:
fs.makedirs(target_directory, exist_ok=True)

for src_path in files_to_copy:
src_path = str(src_path)
filename = Path(src_path).name
dst_path = str(Path(target_directory) / filename)

try:
with fs.open(src_path, "rb") as src_file, fs.open(
dst_path, "wb"
) as dst_file:
dst_file.write(src_file.read())
print(f"Successfully copied {src_path} to {dst_path}")
except Exception as e:
print(f"Error copying {src_path} to {dst_path}: {e}")
return Failure(e)

return Success(None)

except Exception as e:
print(f"Unexpected error during file copying: {e}")
return Failure(e)

0 comments on commit eb0b9f9

Please sign in to comment.