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.

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:

SwiftFileLock

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

Get the upload time of a given path

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

bandersnatch_storage_plugins.swift module

class bandersnatch_storage_plugins.swift.SwiftDirEntry(entry: dict)[source]

Bases: object

is_dir() → bool[source]
is_file() → bool[source]
class bandersnatch_storage_plugins.swift.SwiftFileLock(lock_file: str | PathLike[str], timeout: float = -1, mode: int = 420, thread_local: bool = True, *, blocking: bool = True, is_singleton: bool = False, **kwargs: Any)[source]

Bases: BaseFileLock

Simply watches the existence of the lock file.

property is_locked: bool

A boolean indicating if the lock file is holding the lock currently.

Changed in version 2.0.0: This was previously a method and is now a property.

Type:

return

property path_backend: type[SwiftPath]
class bandersnatch_storage_plugins.swift.SwiftPath(*args: Any)[source]

Bases: Path

BACKEND: SwiftStorage
absolute() → SwiftPath[source]

Return an absolute version of this path by prepending the current working directory. No normalization or symlink resolution is performed.

Use resolve() to get the canonical path to a file.

property backend: SwiftStorage
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: Connection | None = 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_bytes() → bytes[source]

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

read_text(encoding: str | None = None, errors: str | None = None) → str[source]

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

classmethod register_backend(backend: 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: str | None = 'utf-8', errors: str | None = None) → int[source]

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

write_text(data: str, encoding: str | None = 'utf-8', errors: str | None = None, newline: str | None = 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: ConfigParser | None = None, **kwargs: Any)[source]

Bases: StoragePlugin

PATH_BACKEND

alias of SwiftPath

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

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

connection() → Generator[Connection, None, None][source]
copy_file(source: Path | str, dest: Path | str, dest_container: str | None = None) → None[source]

Copy a file from source to dest

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

Copy the contents of a local file to a destination in swift

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

Delete the provided path, recursively if necessary.

property directory: str
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_config_value(config_key: str, *env_keys: Any, default: str | None = None) → str | None[source]
get_container(container: str | None = 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: Path | str) → int[source]

Get the size of a given path in bytes

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

Get the sha256sum of a given path

get_lock(path: str | None = 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:

SwiftFileLock

get_object(container_name: str, file_path: str) → bytes[source]

Retrieve an object from swift, base64 decoding the contents.

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.

Check whether the provided path is a symlink

mkdir(path: 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: Path | str, dest: Path | str, dest_container: str | None = None) → None[source]

Move a file from source to dest

name = 'swift'
open_file(path: 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: 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 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[SwiftDirEntry, 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

Create a symlink at dest that points back at source

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.

update_timestamp(path: Path | str) → None[source]
walk(root: Path | str, dirs: bool = True, conn: Connection | None = None) → list[SwiftPath][source]
write_file(path: Path | str, contents: str | bytes | IO, encoding: str | None = None, errors: str | None = 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).