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
- compare_files(file1: Union[Path, str], file2: Union[Path, str]) bool [source]¶
Compare two files, returning true if they are the same and False if not.
- copy_file(source: Union[Path, str], dest: Union[Path, str]) None [source]¶
Copy a file from source to dest
- delete_file(path: Union[Path, str], dry_run: bool = False) int [source]¶
Delete the provided path, recursively if necessary.
- find(root: Union[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_hash(path: Union[Path, str], function: str = 'sha256') str [source]¶
Get the sha256sum of a given path
- get_lock(path: Optional[str] = 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:
- mkdir(path: Union[Path, str], exist_ok: bool = False, parents: bool = False) None [source]¶
Create the provided directory
- move_file(source: Union[Path, str], dest: Union[Path, str]) None [source]¶
Move a file from source to dest
- name = 'filesystem'¶
- open_file(path: Union[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[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[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[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[Path, str], time: datetime) None [source]¶
Set the upload time of a given path
bandersnatch_storage_plugins.swift module¶
- class bandersnatch_storage_plugins.swift.SwiftFileLock(lock_file: str, timeout: int = -1, backend: Optional[SwiftStorage] = None)[source]¶
Bases:
BaseFileLock
Simply watches the existence of the lock file.
- class bandersnatch_storage_plugins.swift.SwiftPath(*args: Any)[source]¶
Bases:
Path
- BACKEND: SwiftStorage¶
- absolute() 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: SwiftStorage¶
- is_file() bool [source]¶
Whether this path is a regular file (also True for symlinks pointing to regular files).
- iterdir(conn: Optional[Connection] = None, recurse: bool = False, include_swiftkeep: bool = False) Generator[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: SwiftStorage) None [source]¶
- symlink_to(src: Union[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] = None, **kwargs: Any)[source]¶
Bases:
StoragePlugin
- compare_files(file1: Union[Path, str], file2: Union[Path, str]) bool [source]¶
Compare two files, returning true if they are the same and False if not.
- copy_file(source: Union[Path, str], dest: Union[Path, str], dest_container: Optional[str] = None) None [source]¶
Copy a file from source to dest
- copy_local_file(source: Union[Path, str], dest: Union[Path, str]) None [source]¶
Copy the contents of a local file to a destination in swift
- delete_file(path: Union[Path, str], dry_run: bool = False) int [source]¶
Delete the provided path, recursively if necessary.
- find(root: Union[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_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_hash(path: Union[Path, str], function: str = 'sha256') str [source]¶
Get the sha256sum of a given path
- get_lock(path: Optional[str] = None) 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.
- mkdir(path: Union[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[Path, str], dest: Union[Path, str], dest_container: Optional[str] = None) None [source]¶
Move a file from source to dest
- name = 'swift'¶
- open_file(path: Union[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[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[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[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[Path, str], time: datetime) None [source]¶
Set the upload time of a given path
- symlink(src: Union[Path, str], dest: Union[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[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[Path, str], dirs: bool = True, conn: Optional[Connection] = None) List[SwiftPath] [source]¶
- write_file(path: Union[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).