Source code for bandersnatch.package
import asyncio
import logging
from typing import TYPE_CHECKING, Any
from packaging.utils import canonicalize_name
from .errors import ConnectionTimeout, PackageNotFound, StaleMetadata
from .master import StalePage
if TYPE_CHECKING: # pragma: no cover
from .filter import Filter
from .master import Master
logger = logging.getLogger(__name__)
[docs]
class Package:
def __init__(self, name: str, serial: int = 0) -> None:
self.name: str = canonicalize_name(name)
self.raw_name = name
self.serial = serial
self._metadata: dict | None = None
@property
def metadata(self) -> dict[str, Any]:
assert self._metadata is not None, "Must fetch metadata before accessing it"
return self._metadata
@property
def info(self) -> Any:
return self.metadata["info"]
@property
def last_serial(self) -> int:
return int(self.metadata["last_serial"])
@property
def releases(self) -> Any:
return self.metadata["releases"]
@property
def release_files(self) -> list:
release_files: list[dict] = []
for release in self.releases.values():
release_files.extend(release)
return release_files
[docs]
def filter_all_releases(self, release_filters: list["Filter"]) -> bool:
"""
Filter releases and removes releases that fail the filters
"""
releases = list(self.releases.keys())
release_data = {
"info": self.info,
}
pinned_version = False
pinned_plugin = -1
for plugin in release_filters:
pinned_plugin += 1
if plugin.name == "project_requirements_pinned":
if plugin.pinned_version_exists(release_data):
pinned_version = True
break
if pinned_version:
pinned_filter = release_filters[pinned_plugin]
for version in releases:
release_data = {
"version": version,
"releases": self.releases,
"info": self.info,
}
if not pinned_filter.filter(release_data):
del self.releases[version]
else:
for version in releases:
release_data = {
"version": version,
"releases": self.releases,
"info": self.info,
}
if not all(plugin.filter(release_data) for plugin in release_filters):
del self.releases[version]
if releases:
return True
return False
[docs]
def filter_all_releases_files(self, release_file_filters: list["Filter"]) -> bool:
"""
Filter release files and remove empty releases after doing so.
"""
releases = list(self.releases.keys())
for version in releases:
release_files = list(self.releases[version])
for rfindex in reversed(range(len(release_files))):
metadata = {
"info": self.info,
"release": version,
"release_file": self.releases[version][rfindex],
}
if not all(plugin.filter(metadata) for plugin in release_file_filters):
del self.releases[version][rfindex]
if not self.releases[version]:
del self.releases[version]
if releases:
return True
return False