bandersnatch package¶
Submodules¶
bandersnatch.configuration module¶
Module containing classes to access the bandersnatch configuration file
-
class
bandersnatch.configuration.
BandersnatchConfig
(*args: Any, **kwargs: Any)[source]¶ Bases:
object
-
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
-
property
-
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
-
property
-
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
-
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
-
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 modifies the download of specific releases or dist files
-
name
= 'release_plugin'¶
-
-
bandersnatch.filter.
filter_metadata_plugins
() → List[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
() → List[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
() → List[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
() → List[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) → List[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.master module¶
-
class
bandersnatch.master.
Master
(url: str, timeout: float = 10.0, global_timeout: Optional[float] = 18000.0)[source]¶ Bases:
object
-
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
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[source]¶
-
property
xmlrpc_url
¶
-
async
-
exception
bandersnatch.master.
PackageNotFound
[source]¶ Bases:
Exception
We asked for package metadata from PyPI and it wasn’t available
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)[source]¶ Bases:
object
-
async
cleanup_non_pep_503_paths
(package: bandersnatch.package.Package) → 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
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¶
-
property
statusfile
¶
-
stop_on_error
= False¶
-
synced_serial
= 0¶
-
target_serial
= None¶
-
property
todolist
¶
-
property
webdir
¶
-
async
-
async
bandersnatch.mirror.
mirror
(config: configparser.ConfigParser, specific_packages: Optional[List[str]] = None) → int[source]¶
bandersnatch.package module¶
-
class
bandersnatch.package.
Package
(name: str, serial: Union[int, str], mirror: Mirror)[source]¶ Bases:
object
-
async
download_file
(url: str, sha256sum: str, chunk_size: int = 65536) → Optional[pathlib.Path][source]¶
-
property
json_file
¶
-
property
json_pypi_symlink
¶
-
save_json_metadata
(package_info: Dict) → bool[source]¶ Take the JSON metadata we just fetched and save to disk
-
property
simple_directory
¶
-
async
bandersnatch.storage module¶
Storage management
-
class
bandersnatch.storage.
Storage
(*args: Any, config: Optional[configparser.ConfigParser] = None, **kwargs: Any)[source]¶ Bases:
object
Base Storage class
-
PATH_BACKEND
¶ alias of
pathlib.Path
-
compare_files
(file1: Union[pathlib.Path, str], file2: Union[pathlib.Path, str]) → bool[source]¶ 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[source]¶ Copy a file from source to dest
-
delete
(path: Union[pathlib.Path, str], dry_run: bool = False) → int[source]¶ Delete the provided path.
-
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.
-
get_flock_path
() → Union[pathlib.Path, str][source]¶
-
get_hash
(path: Union[pathlib.Path, str], function: str = 'sha256') → str[source]¶ Get the sha256sum of a given path
-
get_json_paths
(name: str) → Sequence[Union[pathlib.Path, str]][source]¶
-
get_lock
(path: str) → filelock.BaseFileLock[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
filelock.BaseFileLock
-
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.
-
iter_dir
(path: Union[pathlib.Path, str]) → Generator[Union[pathlib.Path, str], None, None][source]¶ Iterate over the path, returning the sub-paths
-
mkdir
(path: Union[pathlib.Path, str], exist_ok: bool = False, parents: bool = False) → None[source]¶ Create the provided directory
-
name
= 'storage'¶
-
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 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]¶ 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][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.
-
symlink
(source: Union[pathlib.Path, str], dest: Union[pathlib.Path, str]) → None[source]¶ 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.
-
-
class
bandersnatch.storage.
StoragePlugin
(*args: Any, config: Optional[configparser.ConfigParser] = None, **kwargs: Any)[source]¶ 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][source]¶ 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][source]¶ 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[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.
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: pathlib.Path, 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.
-
bandersnatch.utils.
unlink_parent_dir
(path: pathlib.Path) → None[source]¶ Remove a file and if the dir is empty remove it
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]¶