bandersnatch_storage_plugins package

Package contents

Submodules

bandersnatch_storage_plugins.filesystem module

class bandersnatch_storage_plugins.filesystem.FilesystemStorage(*args: Any, **kwargs: Any)[source]

Bases: StoragePlugin

PATH_BACKEND

alias of Path

compare_files(file1: Path | str, file2: Path | str) bool[source]

Compare two files, returning true if they are the same and False if not.

copy_file(source: Path | str, dest: Path | str) None[source]

Copy a file from source to dest

delete_file(path: Path | str, dry_run: bool = False) int[source]

Delete the provided path, recursively if necessary.

delete_package_file(path: Path | str) None[source]

Remove a file and if the dir is empty remove it

exists(path: Path | str) bool[source]

Check whether the provided path exists

find(root: Path | str, dirs: bool = True) str[source]

A test helper simulating ‘find’.

Iterates over directories and filenames, given as relative paths to the root.

get_file_size(path: Path | str) int[source]

Return the file size of provided path.

get_hash(path: Path | str, function: str = 'sha256') str[source]

Get the sha256sum of a given path

get_lock(path: str | None = None) UnixFileLock[source]

Retrieve the appropriate FileLock backend for this storage plugin

Parameters:

path (str) – The path to use for locking

Returns:

A FileLock backend for obtaining locks

Return type:

filelock.FileLock

get_upload_time(path: Path | str) datetime[source]

Get the upload time of a given path

initialize_plugin() None[source]

Code to initialize the plugin

is_dir(path: Path | str) bool[source]

Check whether the provided path is a directory.

is_file(path: Path | str) bool[source]

Check whether the provided path is a file.

mkdir(path: Path | str, exist_ok: bool = False, parents: bool = False) None[source]

Create the provided directory

move_file(source: Path | str, dest: Path | str) None[source]

Move a file from source to dest

name = 'filesystem'
open_file(path: Path | str, text: bool = True, encoding: str = 'utf-8') Generator[IO, None, None][source]

Yield a file context to iterate over. If text is true, open the file with ‘rb’ mode specified.

read_file(path: Path | str, text: bool = True, encoding: str = 'utf-8', errors: str | None = None) str | bytes[source]

Return the contents of the requested file, either a bytestring or a unicode string depending on whether text is True

rewrite(filepath: Path | str, mode: str = 'w', **kw: Any) Generator[IO, None, None][source]

Rewrite an existing file atomically to avoid programs running in parallel to have race conditions while reading.

rmdir(path: Path | str, recurse: bool = False, force: bool = False, ignore_errors: bool = False, dry_run: bool = False) int[source]

Remove the directory. If recurse is True, allow removing empty children. If force is true, remove contents destructively.

scandir(path: Path | str) Generator[DirEntry, None, None][source]

Read entries from the provided directory

set_upload_time(path: Path | str, time: datetime) None[source]

Set the upload time of a given path

update_safe(filename: Path | str, **kw: Any) Generator[IO, None, None][source]

Rewrite a file atomically.

Clients are allowed to delete the tmpfile to signal that they don’t want to have it updated.

walk(root: Path | str, dirs: bool = True) list[Path][source]
write_file(path: Path | str, contents: str | bytes) None[source]

Write data to the provided path. If contents is a string, the file will be opened and written in “r” + “utf-8” mode, if bytes are supplied it will be accessed using “rb” mode (i.e. binary write).