bandersnatch_storage_plugins package¶
Package contents¶
Submodules¶
bandersnatch_storage_plugins.filesystem module¶
-
class
bandersnatch_storage_plugins.filesystem.
FilesystemStorage
(*args: Any, **kwargs: Any)[source]¶ Bases:
bandersnatch.storage.StoragePlugin
-
PATH_BACKEND
¶ alias of
pathlib.Path
-
compare_files
(file1: Union[pathlib.Path, str], file2: Union[pathlib.Path, str]) → bool[source]¶ Compare two files, returning true if they are the same and False if not.
-
copy_file
(source: Union[pathlib.Path, str], dest: Union[pathlib.Path, str]) → None[source]¶ Copy a file from source to dest
-
delete_file
(path: Union[pathlib.Path, str], dry_run: bool = False) → int[source]¶ Delete the provided path, recursively if necessary.
-
exists
(path: Union[pathlib.Path, str]) → bool[source]¶ Check whether the provided path exists
-
find
(root: Union[pathlib.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: Union[pathlib.Path, str]) → int[source]¶ Return the file size of provided path.
-
get_hash
(path: Union[pathlib.Path, str], function: str = 'sha256') → str[source]¶ Get the sha256sum of a given path
-
get_lock
(path: Optional[str] = None) → filelock.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
-
get_upload_time
(path: Union[pathlib.Path, str]) → datetime.datetime[source]¶ Get the upload time of a given path
-
is_dir
(path: Union[pathlib.Path, str]) → bool[source]¶ Check whether the provided path is a directory.
-
is_file
(path: Union[pathlib.Path, str]) → bool[source]¶ Check whether the provided path is a file.
-
mkdir
(path: Union[pathlib.Path, str], exist_ok: bool = False, parents: bool = False) → None[source]¶ Create the provided directory
-
move_file
(source: Union[pathlib.Path, str], dest: Union[pathlib.Path, str]) → None[source]¶ Move a file from source to dest
-
name
= 'filesystem'¶
-
open_file
(path: Union[pathlib.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: Union[pathlib.Path, str], text: bool = True, encoding: str = 'utf-8', errors: Optional[str] = None) → Union[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: Union[pathlib.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: Union[pathlib.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.
-
set_upload_time
(path: Union[pathlib.Path, str], time: datetime.datetime) → None[source]¶ Set the upload time of a given path
-
update_safe
(filename: Union[pathlib.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: Union[pathlib.Path, str], dirs: bool = True) → List[pathlib.Path][source]¶
-
bandersnatch_storage_plugins.swift module¶
-
class
bandersnatch_storage_plugins.swift.
SwiftFileLock
(lock_file: str, timeout: int = - 1, backend: Optional[bandersnatch_storage_plugins.swift.SwiftStorage] = None)[source]¶ Bases:
filelock.BaseFileLock
Simply watches the existence of the lock file.
-
property
is_locked
¶ True, if the object holds the file lock.
Changed in version 2.0.0: This was previously a method and is now a property.
-
property
path_backend
¶
-
property
-
class
bandersnatch_storage_plugins.swift.
SwiftPath
(*args: Any)[source]¶ Bases:
pathlib.Path
-
absolute
() → bandersnatch_storage_plugins.swift.SwiftPath[source]¶ Return an absolute version of this path. This function works even if the path doesn’t point to anything.
No normalization is done, i.e. all ‘.’ and ‘..’ will be kept along. Use resolve() to get the canonical path to a file.
-
property
backend
¶
-
is_file
() → bool[source]¶ Whether this path is a regular file (also True for symlinks pointing to regular files).
-
iterdir
(conn: Optional[swiftclient.client.Connection] = None, recurse: bool = False, include_swiftkeep: bool = False) → Generator[bandersnatch_storage_plugins.swift.SwiftPath, None, None][source]¶ Iterate over the files in this directory. Does not yield any result for the special paths ‘.’ and ‘..’.
-
mkdir
(mode: int = 511, parents: bool = False, exist_ok: bool = False) → None[source]¶ Create a new directory at this given path.
-
read_text
(encoding: Optional[str] = None, errors: Optional[str] = None) → str[source]¶ Open the file in text mode, read it, and close the file.
-
classmethod
register_backend
(backend: bandersnatch_storage_plugins.swift.SwiftStorage) → None[source]¶
-
symlink_to
(src: Union[pathlib.Path, str], target_is_directory: bool = False, src_container: Optional[str] = None, src_account: Optional[str] = None) → None[source]¶ Make this path a symlink pointing to the given path. Note the order of arguments (self, target) is the reverse of os.symlink’s.
-
unlink
(missing_ok: bool = False) → None[source]¶ Remove this file or link. If the path is a directory, use rmdir() instead.
-
-
class
bandersnatch_storage_plugins.swift.
SwiftStorage
(*args: Any, config: Optional[configparser.ConfigParser] = None, **kwargs: Any)[source]¶ Bases:
bandersnatch.storage.StoragePlugin
-
PATH_BACKEND
¶
-
compare_files
(file1: Union[pathlib.Path, str], file2: Union[pathlib.Path, str]) → bool[source]¶ Compare two files, returning true if they are the same and False if not.
-
copy_file
(source: Union[pathlib.Path, str], dest: Union[pathlib.Path, str], dest_container: Optional[str] = None) → None[source]¶ Copy a file from source to dest
-
copy_local_file
(source: Union[pathlib.Path, str], dest: Union[pathlib.Path, str]) → None[source]¶ Copy the contents of a local file to a destination in swift
-
property
default_container
¶
-
delete_file
(path: Union[pathlib.Path, str], dry_run: bool = False) → int[source]¶ Delete the provided path, recursively if necessary.
-
property
directory
¶
-
exists
(path: Union[pathlib.Path, str]) → bool[source]¶ Check whether the provided path exists
-
find
(root: Union[pathlib.Path, str], dirs: bool = True) → str[source]¶ A test helper simulating ‘find’.
Iterates over directories and filenames, given as relative paths to the root.
-
flock_path
: Union[pathlib.Path, str]¶
-
get_config_value
(config_key: str, *env_keys: Any, default: Optional[str] = None) → Optional[str][source]¶
-
get_container
(container: Optional[str] = None) → List[Dict[str, str]][source]¶ Given the name of a container, return its contents.
- Parameters
container (str) – The name of the desired container, defaults to
default_container
- Returns
A list of objects in the container if it exists
- Return type
Example:
>>> plugin.get_container("bandersnatch") [{ 'bytes': 1101, 'last_modified': '2020-02-27T19:10:17.922970', 'hash': 'a76b4c69bfcf82313bbdc0393b04438a', 'name': 'packages/pyyaml/PyYAML-5.3/LICENSE', 'content_type': 'application/octet-stream' }, { 'bytes': 1779, 'last_modified': '2020-02-27T19:10:17.845520', 'hash': 'c60081e1ad65830b098a7f21a8a8c90e', 'name': 'packages/pyyaml/PyYAML-5.3/PKG-INFO', 'content_type': 'application/octet-stream' }, { 'bytes': 1548, 'last_modified': '2020-02-27T19:10:17.730490', 'hash': '9a8bdf19e93d4b007598b5eb97b461eb', 'name': 'packages/pyyaml/PyYAML-5.3/README', 'content_type': 'application/octet-stream' }, ... ]
-
get_file_size
(path: Union[pathlib.Path, str]) → int[source]¶ Get the size of a given path in bytes
-
get_hash
(path: Union[pathlib.Path, str], function: str = 'sha256') → str[source]¶ Get the sha256sum of a given path
-
get_lock
(path: Optional[str] = None) → bandersnatch_storage_plugins.swift.SwiftFileLock[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
-
get_object
(container_name: str, file_path: str) → bytes[source]¶ Retrieve an object from swift, base64 decoding the contents.
-
get_upload_time
(path: Union[pathlib.Path, str]) → datetime.datetime[source]¶ Get the upload time of a given path
-
is_dir
(path: Union[pathlib.Path, str]) → bool[source]¶ Check whether the provided path is a directory.
-
is_file
(path: Union[pathlib.Path, str]) → bool[source]¶ Check whether the provided path is a file.
-
is_symlink
(path: Union[pathlib.Path, str]) → bool[source]¶ Check whether the provided path is a symlink
-
mkdir
(path: Union[pathlib.Path, str], exist_ok: bool = False, parents: bool = False) → None[source]¶ Create the provided directory
This operation is a no-op on swift.
-
move_file
(source: Union[pathlib.Path, str], dest: Union[pathlib.Path, str], dest_container: Optional[str] = None) → None[source]¶ Move a file from source to dest
-
name
= 'swift'¶
-
open_file
(path: Union[pathlib.Path, str], text: bool = True) → Generator[IO, None, None][source]¶ Yield a file context to iterate over. If text is false, open the file with ‘rb’ mode specified.
-
read_file
(path: Union[pathlib.Path, str], text: bool = True, encoding: str = 'utf-8', errors: Optional[str] = None) → Union[str, bytes][source]¶ Return the contents of the requested file, either a a bytestring or a unicode string depending on whether text is True
-
rewrite
(filepath: Union[pathlib.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: Union[pathlib.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.
-
set_upload_time
(path: Union[pathlib.Path, str], time: datetime.datetime) → None[source]¶ Set the upload time of a given path
-
symlink
(src: Union[pathlib.Path, str], dest: Union[pathlib.Path, str], src_container: Optional[str] = None, src_account: Optional[str] = None) → None[source]¶ Create a symlink at dest that points back at source
-
update_safe
(filename: Union[pathlib.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.
-
update_timestamp
(path: Union[pathlib.Path, str]) → None[source]¶
-
walk
(root: Union[pathlib.Path, str], dirs: bool = True, conn: Optional[swiftclient.client.Connection] = None) → List[bandersnatch_storage_plugins.swift.SwiftPath][source]¶
-
write_file
(path: Union[pathlib.Path, str], contents: Union[str, bytes, IO], encoding: Optional[str] = None, errors: Optional[str] = None) → 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).
-