bandersnatch package

Submodules

bandersnatch.configuration module

Module containing classes to access the bandersnatch configuration file

class bandersnatch.configuration.BandersnatchConfig(config_file: Optional[str] = None)[source]

Bases: object

load_configuration() → None[source]

Read the configuration from a configuration file

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

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

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[source]

Bases: type

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

bandersnatch.delete module

async bandersnatch.delete.delete_packages(config: configparser.ConfigParser, args: argparse.Namespace, master: bandersnatch.master.Master) → int[source]
bandersnatch.delete.delete_path(blob_path: pathlib.Path, dry_run: bool = False) → int[source]

bandersnatch.filter module

Blacklist management

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

Bases: object

Base Filter class

check_match(**kwargs: Any) → bool[source]

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[source]

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[source]

Code to initialize the plugin

name = 'filter'
class bandersnatch.filter.FilterMetadataPlugin(*args: Any, **kwargs: Any)[source]

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

Bases: bandersnatch.filter.Filter

Plugin that blocks sync operations for an entire project

name = 'project_plugin'
class bandersnatch.filter.FilterReleaseFilePlugin(*args: Any, **kwargs: Any)[source]

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

Bases: bandersnatch.filter.Filter

Plugin that modify the download of specific release or dist files

name = 'release_plugin'
bandersnatch.filter.filter_metadata_plugins() → Iterable[bandersnatch.filter.Filter][source]

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.filter.filter_project_plugins() → Iterable[bandersnatch.filter.Filter][source]

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.filter.filter_release_file_plugins() → Iterable[bandersnatch.filter.Filter][source]

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

bandersnatch.filter.filter_release_plugins() → Iterable[bandersnatch.filter.Filter][source]

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.filter.load_filter_plugins(entrypoint_group: str) → Iterable[bandersnatch.filter.Filter][source]

Load all blacklist plugins that are registered with importlib.resources

Parameters

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

Returns

A list of objects derived from the Blacklist class

Return type

List of Blacklist

bandersnatch.log module

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

bandersnatch.main module

async bandersnatch.main.async_main(args: argparse.Namespace, config: configparser.ConfigParser) → int[source]
bandersnatch.main.main(loop: Optional[asyncio.events.AbstractEventLoop] = None) → int[source]

bandersnatch.master module

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

Bases: object

async all_packages() → Dict[str, int][source]
async changed_packages(last_serial: int) → Dict[str, int][source]
async check_for_stale_cache(path: str, required_serial: Optional[int], got_serial: Optional[int]) → None[source]
get(path: str, required_serial: Optional[int], **kw: Any) → AsyncGenerator[aiohttp.client_reqrep.ClientResponse, None][source]
async get_package_metadata(package_name: str, serial: int = 0) → Any[source]
async rpc(method_name: str, serial: int = 0) → Any[source]
async url_fetch(url: str, file_path: pathlib.Path, executor: Union[concurrent.futures.process.ProcessPoolExecutor, concurrent.futures.thread.ThreadPoolExecutor, None] = None, chunk_size: int = 65536) → None[source]
property xmlrpc_url
exception bandersnatch.master.PackageNotFound[source]

Bases: Exception

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

exception bandersnatch.master.StalePage[source]

Bases: Exception

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

exception bandersnatch.master.XmlRpcError[source]

Bases: aiohttp.client_exceptions.ClientError

Issue getting package listing from PyPI Repository

bandersnatch.mirror module

class bandersnatch.mirror.Mirror(homedir: str, 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: Union[pathlib.Path, str, None] = None, diff_append_epoch: bool = False, diff_full_path: Union[pathlib.Path, str, None] = None, flock_timeout: int = 1, diff_file_list: Optional[List] = None, *, cleanup: bool = False)[source]

Bases: object

async determine_packages_to_sync() → None[source]

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][source]

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][source]

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[source]
packages_to_sync: Dict[str, Union[int, str]] = {}
record_finished_package(name: str) → None[source]
root_uri = ''
property statusfile
stop_on_error = False
sync_index_page() → None[source]
async sync_packages() → None[source]
synced_serial = 0
async synchronize() → Dict[str, Set[str]][source]
target_serial = None
property todolist
property webdir
wrapup_successful_sync() → None[source]
async bandersnatch.mirror.mirror(config: configparser.ConfigParser) → int[source]

bandersnatch.package module

class bandersnatch.package.Package(name: str, serial: Union[int, str], mirror: Mirror, *, cleanup: bool = False)[source]

Bases: object

async cleanup_non_pep_503_paths() → None[source]

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

async download_file(url: str, sha256sum: str, chunk_size: int = 65536) → Optional[pathlib.Path][source]
gen_data_requires_python(release: Dict) → str[source]
generate_simple_page() → str[source]
property json_file
property normalized_legacy_simple_directory
property raw_simple_directory
save_json_metadata(package_info: Dict) → bool[source]

Take the JSON metadata we just fetched and save to disk

property simple_directory
sleep_on_stale = 1
async sync(stop_on_error: bool = False, attempts: int = 3) → None[source]
async sync_release_files() → None[source]

Purge + download files returning files removed + added

sync_simple_page() → None[source]
tries = 0

bandersnatch.utils module

bandersnatch.utils.bandersnatch_safe_name(name: str) → str[source]

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[source]
bandersnatch.utils.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.

bandersnatch.utils.hash(path: str, function: str = 'sha256') → str[source]
bandersnatch.utils.make_time_stamp() → str[source]

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[source]
bandersnatch.utils.rewrite(filepath: Union[str, pathlib.Path], 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.

Remove a file and if the dir is empty remove it

bandersnatch.utils.user_agent() → str[source]

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[source]
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[source]
async bandersnatch.verify.metadata_verify(config: configparser.ConfigParser, args: argparse.Namespace) → int[source]

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[source]
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[source]

Module contents