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

SwiftFileLock

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]
write_file(path: Union[pathlib.Path, str], contents: Union[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).

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
class bandersnatch_storage_plugins.swift.SwiftPath(*args: Any)[source]

Bases: pathlib.Path

BACKEND: bandersnatch_storage_plugins.swift.SwiftStorage
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
exists()bool[source]

Whether this path exists.

is_dir()bool[source]

Whether this path is a directory.

is_file()bool[source]

Whether this path is a regular file (also True for symlinks pointing to regular files).

Whether this path is a symbolic link.

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_bytes()bytes[source]

Open the file in bytes mode, read it, and close the file.

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]

Make this path a symlink pointing to the given path. Note the order of arguments (self, target) is the reverse of os.symlink’s.

touch()None[source]

Create this file with the given access mode, if it doesn’t exist.

Remove this file or link. If the path is a directory, use rmdir() instead.

write_bytes(contents: bytes, encoding: Optional[str] = 'utf-8', errors: Optional[str] = None)int[source]

Open the file in bytes mode, write to it, and close the file.

write_text(contents: Optional[str], encoding: Optional[str] = 'utf-8', errors: Optional[str] = None)int[source]

Open the file in text mode, write to it, and close the file.

class bandersnatch_storage_plugins.swift.SwiftStorage(*args: Any, config: Optional[configparser.ConfigParser] = None, **kwargs: Any)[source]

Bases: bandersnatch.storage.StoragePlugin

PATH_BACKEND

alias of bandersnatch_storage_plugins.swift.SwiftPath

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.

connection()Generator[swiftclient.client.Connection, None, None][source]
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

List[Dict[str, str]]

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

SwiftFileLock

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

initialize_plugin()None[source]

Code to initialize the plugin

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.

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

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).