import re
from frozenlist2 import frozenlist
from more_executors.futures import f_map, f_proxy, f_return, f_zip, f_flat_map
from .base import Repository, SyncOptions, repo_type, Importer
from ..attr import pulp_attrib
from ..common import DetachedException
from ...model.unit import RpmUnit
from ... import compat_attr as attr, comps
from ...criteria import Criteria, Matcher
[docs]@attr.s(kw_only=True, frozen=True)
class YumImporter(Importer):
type_id = pulp_attrib(
default="yum_importer", type=str, pulp_field="importer_type_id"
)
"""
Specific importer_type_id for Yum repositories.
"""
[docs]@attr.s(kw_only=True, frozen=True)
class YumSyncOptions(SyncOptions):
"""Options controlling a yum repository
:meth:`~pubtools.pulplib.Repository.sync`.
"""
query_auth_token = pulp_attrib(default=None, type=str)
"""An authorization token that will be added to every request made to the feed URL's server
"""
max_downloads = pulp_attrib(default=None, type=int)
"""Number of threads used when synchronizing the repository.
"""
remove_missing = pulp_attrib(default=None, type=bool)
"""If true, as the repository is synchronized, old rpms will be removed.
"""
retain_old_count = pulp_attrib(default=None, type=int)
"""Count indicating how many old rpm versions to retain.
"""
skip = pulp_attrib(default=None, type=list)
"""List of content types to be skipped during the repository synchronization
"""
checksum_type = pulp_attrib(default=None, type=str)
"""checksum type to use for metadata generation.
Defaults to source checksum type of sha256
"""
num_retries = pulp_attrib(default=None, type=int)
"""Number of times to retry before declaring an error during repository synchronization
Default is 2.
"""
download_policy = pulp_attrib(default=None, type=str)
"""Set the download policy for a repository.
Supported options are immediate,on_demand,background
"""
force_full = pulp_attrib(default=None, type=bool)
"""Boolean flag. If true, full re-sync is triggered.
"""
require_signature = pulp_attrib(default=None, type=bool)
"""Requires that imported packages like RPM/DRPM/SRPM should be signed
"""
allowed_keys = pulp_attrib(default=None, type=list)
"""List of allowed signature key IDs that imported packages can be signed with
"""
[docs]@repo_type("rpm-repo")
@attr.s(kw_only=True, frozen=True)
class YumRepository(Repository):
"""A :class:`~pubtools.pulplib.Repository` for RPMs, errata and related content."""
# this class only overrides some defaults for attributes defined in super
type = pulp_attrib(default="rpm-repo", type=str, pulp_field="notes._repo-type")
population_sources = pulp_attrib(
default=attr.Factory(frozenlist),
type=list,
converter=frozenlist,
pulp_field="notes.population_sources",
)
"""List of repository IDs used to populate this repository
"""
ubi_population = pulp_attrib(
default=False, type=bool, pulp_field="notes.ubi_population"
)
"""Flag indicating whether repo should be populated from population_sources for the purposes of UBI
"""
mutable_urls = attr.ib(
default=attr.Factory(lambda: frozenlist(["repodata/repomd.xml"])),
type=list,
converter=frozenlist,
)
ubi_config_version = pulp_attrib(
default=None, type=str, pulp_field="notes.ubi_config_version"
)
"""Version of UBI config that should be used for population of this repository."""
importer = pulp_attrib(
default=YumImporter(),
type=YumImporter,
pulp_field="importers",
pulp_py_converter=YumImporter._from_data,
py_pulp_converter=YumImporter._to_data,
)
"""
An object of :class:`~pubtools.pulplib.YumImporter` that is associated with the repository.
.. versionadded:: 2.39.0
"""
[docs] def get_binary_repository(self):
"""Find and return the binary repository relating to this repository.
Yum repositories usually come in triplets of
(binary RPMs, debuginfo RPMs, source RPMs). For example:
.. list-table::
:widths: 75 25
* - ``rhel-7-server-rpms__7Server__x86_64``
- binary
* - ``rhel-7-server-debug-rpms__7Server__x86_64``
- debug
* - ``rhel-7-server-source-rpms__7Server__x86_64``
- source
This method along with :meth:`get_debug_repository` and :meth:`get_source_repository` allow locating other repositories
from within this group.
Returns:
``Future[YumRepository]``
Binary repository relating to this repository.
``Future[None]``
If there is no related repository.
"""
return self._get_related_repository(repo_t="binary")
[docs] def get_debug_repository(self):
"""Find and return the debug repository relating to this repository.
Returns:
``Future[YumRepository]``
Debug repository relating to this repository.
``Future[None]``
If there is no related repository.
"""
return self._get_related_repository(repo_t="debug")
[docs] def get_source_repository(self):
"""Find and return the source repository relating to this repository.
Returns:
``Future[YumRepository]``
Source repository relating to this repository.
``Future[None]``
If there is no related repository.
"""
return self._get_related_repository(repo_t="source")
def _get_related_repository(self, repo_t):
if not self._client:
raise DetachedException()
suffixes_mapping = {
"binary": "/os",
"debug": "/debug",
"source": "/source/SRPMS",
}
regex = r"(/os|/source/SRPMS|/debug)$"
def unpack_page(page):
if len(page.data) != 1:
return None
return page.data[0]
suffix = suffixes_mapping[repo_t]
if str(self.relative_url).endswith(suffix):
return f_proxy(f_return(self))
base_url = re.sub(regex, "", self.relative_url)
relative_url = base_url + suffix
criteria = Criteria.and_(
Criteria.with_field("relative_url", relative_url),
Criteria.with_field("type_id", "yum_distributor"),
)
page_f = self._client.search_distributor(criteria)
distributor_f = f_proxy(f_map(page_f, unpack_page))
if distributor_f.result() is None:
out = f_return()
else:
out = self._client.get_repository(distributor_f.repo_id)
return out
[docs] def upload_rpm(self, file_obj, **kwargs):
"""Upload an RPM to this repository.
.. warning::
For RPMs belonging to a module, it's strongly advised to upload
the module metadata first (using :meth:`upload_modules`) and only
proceed with uploading RPMs once module upload has completed.
This reduces the risk of accidentally publishing a repository with
modular RPMs without the corresponding metadata (which has a much
worse impact than publishing metadata without the corresponding RPMs).
Args:
file_obj (str, file object)
If it's a string, then it's the path of an RPM to upload.
Otherwise, it should be a
`file-like object <https://docs.python.org/3/glossary.html#term-file-object>`_
pointing at the bytes to upload.
The client takes ownership of this file object; it should
not be modified elsewhere, and will be closed when upload
completes.
kwargs
Additional field values to set on the uploaded unit.
Any :class:`~pubtools.pulplib.RpmUnit` fields documented as
*mutable* may be included here (for example, ``cdn_path``).
An error will occur if attempting to set other fields.
Returns:
Future[list of :class:`~pubtools.pulplib.Task`]
A future which is resolved after content has been imported
to this repo.
Raises:
DetachedException
If this instance is not attached to a Pulp client.
.. versionadded:: 2.16.0
.. versionadded:: 2.20.0
Added ability to set mutable fields on upload.
"""
# We want some name of what we're uploading for logging purposes, but the
# input could be a plain string, or a file object with 'name' attribute, or
# a file object without 'name' ... make sure we do something reasonable in
# all cases.
if isinstance(file_obj, str):
name = file_obj
else:
# If we don't know what we're uploading we just say it's "an RPM"...
name = getattr(file_obj, "name", "an RPM")
unit_metadata_fn = None
usermeta = RpmUnit._usermeta_from_kwargs(**kwargs)
if usermeta:
unit_metadata_fn = lambda _: usermeta
return self._upload_then_import(
file_obj, name, "rpm", unit_metadata_fn=unit_metadata_fn
)
[docs] def upload_modules(self, file_obj):
"""Upload a modulemd stream to this repository.
All supported documents in the given stream will be imported to this
repository. On current versions of Pulp 2.x, this means only:
* `modulemd v2 <https://github.com/fedora-modularity/libmodulemd/blob/main/yaml_specs/modulemd_stream_v2.yaml>`_
* `modulemd-defaults v1 <https://github.com/fedora-modularity/libmodulemd/blob/main/yaml_specs/modulemd_defaults_v1.yaml>`_
Attempting to use other document types may result in an error.
Args:
file_obj (str, file object)
If it's a string, then it's the path of a modulemd YAML
file to upload.
Otherwise, it should be a
`file-like object <https://docs.python.org/3/glossary.html#term-file-object>`_
pointing at the text to upload.
The client takes ownership of this file object; it should
not be modified elsewhere, and will be closed when upload
completes.
Returns:
Future[list of :class:`~pubtools.pulplib.Task`]
A future which is resolved after content has been imported
to this repo.
Raises:
DetachedException
If this instance is not attached to a Pulp client.
.. versionadded:: 2.17.0
"""
if isinstance(file_obj, str):
name = file_obj
else:
name = getattr(file_obj, "name", "modulemds")
return self._upload_then_import(file_obj, name, "modulemd")
[docs] def upload_comps_xml(self, file_obj):
"""Upload a comps.xml file to this repository.
.. warning::
Beware of the following quirks with respect to the upload of comps.xml:
* Pulp does not directly store the uploaded XML. Instead, this library
parses the XML and uses the content to store various units. The comps
XML rendered as a yum repository is published is therefore not
guaranteed to be bytewise-identical to the uploaded content.
* The uploaded XML must contain all comps data for the repo, as
any existing comps data will be removed from the repo.
* The XML parser is not secure against maliciously constructed data.
* The process of parsing the XML and storing units consists of multiple
steps which cannot be executed atomically. That means *if this
operation is interrupted, the repository may be left with incomplete
data*. It's recommended to avoid publishing a repository in this state.
Args:
file_obj (str, file object)
If it's a string, then it's the path of a comps XML
file to upload.
Otherwise, it should be a
`file-like object <https://docs.python.org/3/glossary.html#term-file-object>`_
pointing at the bytes of a valid comps.xml file.
The client takes ownership of this file object; it should
not be modified elsewhere, and will be closed when upload
completes.
Returns:
Future[list of :class:`~pubtools.pulplib.Task`]
A future which is resolved after content has been imported
to this repo.
Raises:
DetachedException
If this instance is not attached to a Pulp client.
.. versionadded:: 2.17.0
"""
if isinstance(file_obj, str):
file_name = file_obj
file_obj = open(file_obj, "rb")
else:
file_name = getattr(file_obj, "name", "comps.xml")
# Parse the provided XML. We will crash here if the given XML is not
# valid.
with file_obj:
unit_dicts = comps.units_for_xml(file_obj)
# Every comps-related unit type has a repo_id which should reference the repo
# we're uploading to.
for unit in unit_dicts:
unit["repo_id"] = self.id
comps_type_ids = [
"package_group",
"package_category",
"package_environment",
"package_langpacks",
]
# Remove former units of comps-related types so that the end result is only
# those units included in the current XML.
crit = Criteria.with_field("content_type_id", Matcher.in_(comps_type_ids))
out = self.remove_content(crit)
# Once removal is done we can upload each unit.
upload_f = []
for unit_dict in unit_dicts:
type_id = unit_dict["_content_type_id"]
# For one comps.xml we are doing multiple upload operations, each of
# which would be logged independently. Come up with some reasonable name
# for each unit to put into the logs.
#
# Example: if uploading my-comps.xml and processing a package_group
# with id kde-desktop-environment, the name for logging purposes would
# be: "my-comps.xml [group.kde-desktop-environment]".
#
unit_name = type_id.replace("package_", "")
if unit_dict.get("id"):
unit_name = "%s.%s" % (unit_name, unit_dict["id"])
unit_name = "%s [%s]" % (file_name, unit_name)
upload_f.append(
f_flat_map(
out, self._comps_unit_uploader(unit_name, type_id, unit_dict)
)
)
# If there were no units to upload then just return the removal.
if not upload_f:
return out
# There were uploads, then we'll wait for all of them to complete and
# return the tasks for all.
out = f_zip(*upload_f)
out = f_map(out, lambda uploads: sum(uploads, []))
return out
def _comps_unit_uploader(self, name, type_id, metadata):
# A helper used from upload_comps_xml.
#
# This helper only exists to eagerly bind arguments for a single
# unit upload, due to confusing behavior around variable scope
# when combining loops and lambdas.
def upload(_unused):
return self._upload_then_import(
file_obj=None,
name=name,
type_id=type_id,
unit_metadata_fn=lambda _: metadata,
)
return upload
[docs] def upload_erratum(self, erratum):
"""Upload an erratum/advisory object to this repository.
.. warning::
There are many quirks with respect to advisory upload. Please be aware
of the following before using this API:
* Only one advisory with a given ``id`` may exist in the system.
* When uploading an advisory with an ``id`` equal to one already in the
system, the upload will generally be ignored (i.e. complete successfully
but have no effect), unless either the ``version`` or ``updated`` fields
have a value larger than the existing advisory.
This implies that, if you want to ensure an existing advisory is updated,
you must first search for the existing object and mutate one of these
fields before uploading a modified object. *The library will not take
care of this for you.*
* When overwriting an existing advisory, all fields will be overwritten.
The sole exception is the ``pkglist`` field which will be merged with
existing data when applicable.
* If an advisory with the same ``id`` is present in multiple published yum
repositories with inconsistent fields, yum/dnf client errors or warnings
may occur. It's therefore recommended that, whenever an existing
advisory is modified, every repository containing that advisory should
be republished. *The library will not take care of this for you.*
* The ``repository_memberships`` field on the provided object has no effect
(it cannot be used to upload an advisory to multiple repos at once).
Args:
erratum (:class:`~pubtools.pulplib.ErratumUnit`)
An erratum object.
Unlike most other uploaded content, errata are not backed by any
file; any arbitrarily constructed ErratumUnit may be uploaded.
Returns:
Future[list of :class:`~pubtools.pulplib.Task`]
A future which is resolved after content has been imported
to this repo.
Raises:
DetachedException
If this instance is not attached to a Pulp client.
.. versionadded:: 2.17.0
"""
# Convert from ErratumUnit to a raw Pulp-style dict, recursively.
erratum_dict = erratum._to_data()
# Drop this one field because the _content_type_id, though embedded
# in unit dicts on read, is passed as a separate parameter on write.
type_id = erratum_dict.pop("_content_type_id")
# Drop this because the caller cannot influence the _id (unit id)
# for uploaded units.
del erratum_dict["_id"]
# And drop this one because repository_memberships is synthesized when
# Pulp renders units, and can't be set during import.
del erratum_dict["repository_memberships"]
return self._upload_then_import(
file_obj=None,
name=erratum_dict["id"],
type_id=type_id,
unit_key_fn=lambda _: {"id": erratum_dict["id"]},
unit_metadata_fn=lambda _: erratum_dict,
)
