bandersnatch package

Package contents

Submodules

bandersnatch.configuration module

Module containing classes to access the bandersnatch configuration file

class bandersnatch.configuration.BandersnatchConfig(*args: Any, **kwargs: Any)

Bases: object

load_configuration() → None

Read the configuration from a configuration file

class bandersnatch.configuration.DeprecatedKey(old_section, old_key, new_section, new_key, deprecated_version)

Bases: tuple

property deprecated_version

Alias for field number 4

property new_key

Alias for field number 3

property new_section

Alias for field number 2

property old_key

Alias for field number 1

property old_section

Alias for field number 0

class bandersnatch.configuration.SetConfigValues(json_save, root_uri, diff_file_path, diff_append_epoch, digest_name, storage_backend_name, cleanup)

Bases: tuple

property cleanup

Alias for field number 6

property diff_append_epoch

Alias for field number 3

property diff_file_path

Alias for field number 2

property digest_name

Alias for field number 4

property json_save

Alias for field number 0

property root_uri

Alias for field number 1

property storage_backend_name

Alias for field number 5

class bandersnatch.configuration.Singleton

Bases: type

bandersnatch.configuration.validate_config_values(config: configparser.ConfigParser) → bandersnatch.configuration.SetConfigValues

bandersnatch.delete module

async bandersnatch.delete.delete_packages(config: configparser.ConfigParser, args: argparse.Namespace, master: bandersnatch.master.Master) → int

bandersnatch.delete.delete_path(blob_path: pathlib.Path, dry_run: bool = False) → int

bandersnatch.filter module

Blacklist management

class bandersnatch.filter.Filter(*args: Any, **kwargs: Any)

Bases: object

Base Filter class

check_match(**kwargs: Any) → bool

Check if the plugin matches based on the arguments provides.

Returns

True if the values match a filter rule, False otherwise

Return type

bool

filter(metadata: dict) → bool

Check if the plugin matches based on the package’s metadata.

Returns

True if the values match a filter rule, False otherwise

Return type

bool

initialize_plugin() → None

Code to initialize the plugin

name = 'filter'

class bandersnatch.filter.FilterMetadataPlugin(*args: Any, **kwargs: Any)

Bases: bandersnatch.filter.Filter

Plugin that blocks sync operations for an entire project based on info fields.

name = 'metadata_plugin'

class bandersnatch.filter.FilterProjectPlugin(*args: Any, **kwargs: Any)

Bases: bandersnatch.filter.Filter

Plugin that blocks sync operations for an entire project

name = 'project_plugin'

class bandersnatch.filter.FilterReleaseFilePlugin(*args: Any, **kwargs: Any)

Bases: bandersnatch.filter.Filter

Plugin that modify the download of specific release or dist files

name = 'release_file_plugin'

class bandersnatch.filter.FilterReleasePlugin(*args: Any, **kwargs: Any)

Bases: bandersnatch.filter.Filter

Plugin that modifies the download of specific releases or dist files

name = 'release_plugin'

class bandersnatch.filter.LoadedFilters(load_all: bool = False)

Bases: object

A class to load all of the filters enabled

ENTRYPOINT_GROUPS = ['bandersnatch_filter_plugins.v2.project', 'bandersnatch_filter_plugins.v2.metadata', 'bandersnatch_filter_plugins.v2.release', 'bandersnatch_filter_plugins.v2.release_file']

filter_metadata_plugins() → List[bandersnatch.filter.Filter]

Load and return the release filtering plugin objects

Returns

List of objects derived from the bandersnatch.filter.Filter class

Return type

list of bandersnatch.filter.Filter

filter_project_plugins() → List[bandersnatch.filter.Filter]

Load and return the release filtering plugin objects

Returns

List of objects derived from the bandersnatch.filter.Filter class

Return type

list of bandersnatch.filter.Filter

filter_release_file_plugins() → List[bandersnatch.filter.Filter]

Load and return the release file filtering plugin objects

Returns

List of objects derived from the bandersnatch.filter.Filter class

Return type

list of bandersnatch.filter.Filter

filter_release_plugins() → List[bandersnatch.filter.Filter]

Load and return the release filtering plugin objects

Returns

List of objects derived from the bandersnatch.filter.Filter class

Return type

list of bandersnatch.filter.Filter

bandersnatch.log module

bandersnatch.log.setup_logging(args: Any) → logging.StreamHandler

bandersnatch.main module

async bandersnatch.main.async_main(args: argparse.Namespace, config: configparser.ConfigParser) → int

bandersnatch.main.main(loop: Optional[asyncio.events.AbstractEventLoop] = None) → int

bandersnatch.master module

class bandersnatch.master.Master(url: str, timeout: float = 10.0, global_timeout: Optional[float] = 18000.0)

Bases: object

async all_packages() → Dict[str, int]

async changed_packages(last_serial: int) → Dict[str, int]

async check_for_stale_cache(path: str, required_serial: Optional[int], got_serial: Optional[int]) → None

get(path: str, required_serial: Optional[int], **kw: Any) → AsyncGenerator[aiohttp.client_reqrep.ClientResponse, None]

async get_package_metadata(package_name: str, serial: int = 0) → Any

async rpc(method_name: str, serial: int = 0) → Any

async url_fetch(url: str, file_path: pathlib.Path, executor: Optional[Union[concurrent.futures.process.ProcessPoolExecutor, concurrent.futures.thread.ThreadPoolExecutor]] = None, chunk_size: int = 65536) → None

property xmlrpc_url

exception bandersnatch.master.PackageNotFound(package_name: str)

Bases: Exception

We asked for package metadata from PyPI and it wasn’t available

exception bandersnatch.master.StalePage

Bases: Exception

We got a page back from PyPI that doesn’t meet our expected serial.

exception bandersnatch.master.XmlRpcError

Bases: aiohttp.client_exceptions.ClientError

Issue getting package listing from PyPI Repository

bandersnatch.mirror module

class bandersnatch.mirror.Mirror(homedir: pathlib.Path, master: bandersnatch.master.Master, storage_backend: Optional[str] = None, stop_on_error: bool = False, workers: int = 3, hash_index: bool = False, json_save: bool = False, digest_name: Optional[str] = None, root_uri: Optional[str] = None, keep_index_versions: int = 0, diff_file: Optional[Union[pathlib.Path, str]] = None, diff_append_epoch: bool = False, diff_full_path: Optional[Union[pathlib.Path, str]] = None, flock_timeout: int = 1, diff_file_list: Optional[List] = None, *, cleanup: bool = False)

Bases: object

async cleanup_non_pep_503_paths(package: bandersnatch.package.Package) → None

Before 4.0 we use to store backwards compatible named dirs for older pip This function checks for them and cleans them up

async determine_packages_to_sync() → None

Update the self.packages_to_sync to contain packages that need to be synced.

diff_append_epoch = False

diff_file = None

diff_full_path = None

digest_name = 'sha256'

errors = False

find_package_indexes_in_dir(simple_dir: pathlib.Path) → List[str]

Given a directory that contains simple packages indexes, return a sorted list of normalized package names. This presumes every directory within is a simple package index directory.

property generationfile

get_simple_dirs(simple_dir: pathlib.Path) → List[pathlib.Path]

Return a list of simple index directories that should be searched for package indexes when compiling the main index page.

json_save = False

need_index_sync = True

now = None

async package_syncer(idx: int) → None

packages_to_sync: Dict[str, Union[int, str]] = {}

record_finished_package(name: str) → None

root_uri: Optional[str] = ''

property statusfile

stop_on_error = False

sync_index_page() → None

async sync_packages() → None

synced_serial = 0

async synchronize(specific_packages: Optional[List[str]] = None) → Dict[str, Set[str]]

target_serial = None

property todolist

property webdir

wrapup_successful_sync() → None

async bandersnatch.mirror.mirror(config: configparser.ConfigParser, specific_packages: Optional[List[str]] = None) → int

bandersnatch.package module

class bandersnatch.package.Package(name: str, serial: Union[int, str], mirror: Mirror)

Bases: object

async download_file(url: str, sha256sum: str, chunk_size: int = 65536) → Optional[pathlib.Path]

gen_data_requires_python(release: Dict) → str

generate_simple_page() → str

property info

property json_file

property json_pypi_symlink

property last_serial

property metadata

property release_files

property releases

save_json_metadata(package_info: Dict) → bool

Take the JSON metadata we just fetched and save to disk

property simple_directory

async sync(filters: LoadedFilters, attempts: int = 3) → None

async sync_release_files() → None

Purge + download files returning files removed + added

sync_simple_page() → None

async update_metadata(attempts: int = 3) → None

exception bandersnatch.package.StaleMetadata(package_name: str, attempts: int)

Bases: Exception

We attempted to retreive metadata from PyPI, but it was stale.

bandersnatch.storage module

Storage management

class bandersnatch.storage.Storage(*args: Any, config: Optional[configparser.ConfigParser] = None, **kwargs: Any)

Bases: object

Base Storage class

PATH_BACKEND

alias of pathlib.Path

static canonicalize_package(name: str) → str

compare_files(file1: Union[pathlib.Path, str], file2: Union[pathlib.Path, str]) → bool

Compare two files and determine whether they contain the same data. Return True if they match

copy_file(source: Union[pathlib.Path, str], dest: Union[pathlib.Path, str]) → None

Copy a file from source to dest

delete(path: Union[pathlib.Path, str], dry_run: bool = False) → int

Delete the provided path.

delete_file(path: Union[pathlib.Path, str], dry_run: bool = False) → int

Delete the provided path, recursively if necessary.

property directory

exists(path: Union[pathlib.Path, str]) → bool

Check whether the provided path exists

find(root: Union[pathlib.Path, str], dirs: bool = True) → str

A test helper simulating ‘find’.

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

get_flock_path() → Union[pathlib.Path, str]

get_hash(path: Union[pathlib.Path, str], function: str = 'sha256') → str

Get the sha256sum of a given path

get_json_paths(name: str) → Sequence[Union[pathlib.Path, str]]

get_lock(path: str) → filelock.BaseFileLock

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

hash_file(path: Union[pathlib.Path, str], function: str = 'sha256') → str

initialize_plugin() → None

Code to initialize the plugin

is_dir(path: Union[pathlib.Path, str]) → bool

Check whether the provided path is a directory.

is_file(path: Union[pathlib.Path, str]) → bool

Check whether the provided path is a file.

iter_dir(path: Union[pathlib.Path, str]) → Generator[Union[pathlib.Path, str], None, None]

Iterate over the path, returning the sub-paths

mkdir(path: Union[pathlib.Path, str], exist_ok: bool = False, parents: bool = False) → None

Create the provided directory

name = 'storage'

open_file(path: Union[pathlib.Path, str], text: bool = True) → Generator[IO, None, None]

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]

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

rewrite(filepath: Union[pathlib.Path, str], mode: str = 'w', **kw: Any) → Generator[IO, None, None]

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

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

symlink(source: Union[pathlib.Path, str], dest: Union[pathlib.Path, str]) → None

Create a symlink at dest that points back at source

update_safe(filename: Union[pathlib.Path, str], **kw: Any) → Generator[IO, None, None]

Rewrite a file atomically.

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

write_file(path: Union[pathlib.Path, str], contents: Union[str, bytes]) → None

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

class bandersnatch.storage.StoragePlugin(*args: Any, config: Optional[configparser.ConfigParser] = None, **kwargs: Any)

Bases: bandersnatch.storage.Storage

Plugin that provides a storage backend for bandersnatch

name = 'storage_plugin'

bandersnatch.storage.load_storage_plugins(entrypoint_group: str, enabled_plugin: Optional[str] = None, config: Optional[configparser.ConfigParser] = None, clear_cache: bool = False) → Set[bandersnatch.storage.Storage]

Load all storage plugins that are registered with pkg_resources

Parameters

• entrypoint_group (str) – The entrypoint group name to load plugins from

• enabled_plugin (str) – The optional enabled storage plugin to search for

• config (configparser.ConfigParser) – The optional configparser instance to pass in

• clear_cache (bool) – Whether to clear the plugin cache

Returns

A list of objects derived from the Storage class

Return type

List of Storage

bandersnatch.storage.storage_backend_plugins(backend: Optional[str] = 'filesystem', config: Optional[configparser.ConfigParser] = None, clear_cache: bool = False) → Iterable[bandersnatch.storage.Storage]

Load and return the release filtering plugin objects

Parameters

• backend (str) – The optional enabled storage plugin to search for

• config (configparser.ConfigParser) – The optional configparser instance to pass in

• clear_cache (bool) – Whether to clear the plugin cache

Returns

List of objects derived from the bandersnatch.storage.Storage class

Return type

list of bandersnatch.storage.Storage

bandersnatch.utils module

bandersnatch.utils.bandersnatch_safe_name(name: str) → str

Convert an arbitrary string to a standard distribution name Any runs of non-alphanumeric/. characters are replaced with a single ‘-‘.

• This was copied from pkg_resources (part of setuptools)

bandersnatch also lower cases the returned name

bandersnatch.utils.convert_url_to_path(url: str) → str

bandersnatch.utils.find(root: Union[pathlib.Path, str], dirs: bool = True) → str

A test helper simulating ‘find’.

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

bandersnatch.utils.hash(path: pathlib.Path, function: str = 'sha256') → str

bandersnatch.utils.make_time_stamp() → str

Helper function that returns a timestamp suitable for use in a filename on any OS

bandersnatch.utils.recursive_find_files(files: Set[pathlib.Path], base_dir: pathlib.Path) → None

bandersnatch.utils.rewrite(filepath: Union[str, pathlib.Path], mode: str = 'w', **kw: Any) → Generator[IO, None, None]

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

bandersnatch.utils.unlink_parent_dir(path: pathlib.Path) → None

Remove a file and if the dir is empty remove it

bandersnatch.utils.user_agent() → str

bandersnatch.verify module

async bandersnatch.verify.delete_unowned_files(mirror_base: pathlib.Path, executor: concurrent.futures.thread.ThreadPoolExecutor, all_package_files: List[pathlib.Path], dry_run: bool) → int

async bandersnatch.verify.get_latest_json(master: bandersnatch.master.Master, json_path: pathlib.Path, config: configparser.ConfigParser, executor: Optional[concurrent.futures.thread.ThreadPoolExecutor] = None, delete_removed_packages: bool = False) → None

async bandersnatch.verify.metadata_verify(config: configparser.ConfigParser, args: argparse.Namespace) → int

Crawl all saved JSON metadata or online to check we have all packages if delete - generate a diff of unowned files

async bandersnatch.verify.verify(master: bandersnatch.master.Master, config: configparser.ConfigParser, json_file: str, mirror_base_path: pathlib.Path, all_package_files: List[pathlib.Path], args: argparse.Namespace, executor: Optional[concurrent.futures.thread.ThreadPoolExecutor] = None, releases_key: str = 'releases') → None

async bandersnatch.verify.verify_producer(master: bandersnatch.master.Master, config: configparser.ConfigParser, all_package_files: List[pathlib.Path], mirror_base_path: pathlib.Path, json_files: List[str], args: argparse.Namespace, executor: Optional[concurrent.futures.thread.ThreadPoolExecutor] = None) → None