pulp_glue.common.context¶

def echo(self, message: str, nl: bool = True, err: bool = False) -> None:
    """
    Abstract function that will be called to emit warnings and task progress.

    Warning:
        This function needs to be implemented.
    """
    raise NotImplementedError("PulpContext is an abstract class.")

def prompt(self, text: str, hide_input: bool = False) -> str:
    """
    Abstract function that will be called to ask for a password interactively.

    Note:
        If a password is provided non-interactively, this function need not be implemented.
    """
    raise NotImplementedError("PulpContext is an abstract class.")

def call(
    self,
    operation_id: str,
    non_blocking: bool = False,
    parameters: t.Optional[t.Dict[str, t.Any]] = None,
    body: t.Optional[EntityDefinition] = None,
    validate_body: bool = True,
) -> t.Any:
    """
    Perform an API call for operation_id.
    Wait for triggered tasks to finish if not background.
    Returns the operation result, or the finished task.

    Parameters:
        operation_id: The operation ID in the openapi v3 spec to be called.
        non_blocking: returns unfinished tasks if `True`.
        parameters: Arguments that are to be sent as headers, querystrings or part of the URI.
        body: Body payload for POST, PUT, PATCH calls.
        validate_body: Indicate whether the body should be validated.

    Returns:
        The body of the response, or the task or task group if one was issued.

    Raises:
        PulpNoWait: in case the context has `background_tasks` set or a task (group) timed out.
    """
    if parameters is None:
        parameters = {}
    if self.domain_enabled:
        # Validation will fail if path doesn't need domain parameter
        if "pulp_domain" in self.api.param_spec(operation_id, "path", required=True):
            parameters["pulp_domain"] = self.pulp_domain
    parameters = preprocess_payload(parameters)
    if body is not None:
        body = preprocess_payload(body)
    try:
        result = self.api.call(
            operation_id,
            parameters=parameters,
            body=body,
            validate_body=validate_body,
        )
    except OpenAPIError as e:
        raise PulpException(str(e))
    except HTTPError as e:
        if e.response is not None:
            raise PulpHTTPError(str(e.response.text), e.response.status_code)
        else:
            raise PulpException(str(e))
    # Asynchronous tasks seem to be reported by a dict containing only one key "task"
    if isinstance(result, dict) and ["task"] == list(result.keys()):
        task_href = result["task"]
        result = self.api.call("tasks_read", parameters={"task_href": task_href})
        self.echo(
            _("Started background task {task_href}").format(task_href=task_href), err=True
        )
        if not non_blocking:
            result = self.wait_for_task(result)
    if isinstance(result, dict) and ["task_group"] == list(result.keys()):
        task_group_href = result["task_group"]
        result = self.api.call(
            "task_groups_read", parameters={"task_group_href": task_group_href}
        )
        self.echo(
            _("Started background task group {task_group_href}").format(
                task_group_href=task_group_href
            ),
            err=True,
        )
        if not non_blocking:
            result = self.wait_for_task_group(result)
    return result

def wait_for_task(self, task: EntityDefinition, expect_cancel: bool = False) -> t.Any:
    """
    Wait for a task to finish and return the finished task object.

    Parameters:
        task: A task object to monitor.
        expect_cancel: Swaps the raising condition for completed and canceled tasks.

    Raises:
        PulpNoWait: on timeout or if the context has `background_tasks` set.
        PulpException: on ctrl-c, if task failed or was canceled.
    """
    deadline = datetime.datetime.now() + self.timeout if self.timeout else None

    if self.background_tasks:
        raise PulpNoWait(_("Not waiting for task because --background was specified."))
    task_href = task["pulp_href"]
    try:
        while not self._task_finished(task, expect_cancel=expect_cancel):
            if deadline and datetime.datetime.now() > deadline:
                raise PulpNoWait(
                    _("Waiting for task {task_href} timed out.").format(
                        task_href=task["pulp_href"]
                    )
                )
            time.sleep(1)
            self.echo(".", nl=False, err=True)
            task = self.api.call("tasks_read", parameters={"task_href": task["pulp_href"]})
        self.echo("Done.", err=True)
        return task
    except KeyboardInterrupt:
        raise PulpNoWait(_("Task {task_href} sent to background.").format(task_href=task_href))

def wait_for_task_group(self, task_group: EntityDefinition) -> t.Any:
    """
    Wait for a task group to finish and return the finished task object.

    Parameters:
        task_group: A task group object all of which's tasks to monitor.

    Raises:
        PulpNoWait: on timeout or if the context has `background_tasks` set.
        PulpException: on ctrl-c, if task failed or was canceled.
    """
    deadline = datetime.datetime.now() + self.timeout if self.timeout else None

    if self.background_tasks:
        raise PulpNoWait("Not waiting for task group because --background was specified.")
    try:
        while not self._task_group_finished(task_group):
            if deadline and datetime.datetime.now() > deadline:
                raise PulpNoWait(
                    _("Waiting for task group {task_group_href} timed out.").format(
                        task_group_href=task_group["pulp_href"]
                    )
                )
            time.sleep(1)
            self.echo(".", nl=False, err=True)
            task_group = self.api.call(
                "task_groups_read", parameters={"task_group_href": task_group["pulp_href"]}
            )
    except KeyboardInterrupt:
        raise PulpNoWait(
            _("Task group {task_group_href} sent to background.").format(
                task_group_href=task_group["pulp_href"]
            )
        )

def call(
    self,
    operation: str,
    non_blocking: bool = False,
    parameters: t.Optional[t.Dict[str, t.Any]] = None,
    body: t.Optional[EntityDefinition] = None,
    validate_body: bool = True,
) -> t.Any:
    """
    Perform an API call for operation.
    Wait for triggered tasks to finish if not background.
    Returns the operation result, or the finished task.

    Parameters:
        operation: The operation to be performed on the entity. Usually the openapi
            operation_id is constructed by concatenating with the `ID_PREFIX`.
        non_blocking: returns unfinished tasks if `True`.
        parameters: Arguments that are to be sent as headers, querystrings or part of the URI.
        body: Body payload for POST, PUT, PATCH calls.
        validate_body: Indicate whether the body should be validated.

    Returns:
        The body of the response, or the task or task group if one was issued.

    Raises:
        PulpNoWait: in case the context has `background_tasks` set or a task (group) timed out.
    """
    operation_id: str = (
        getattr(self, operation.upper() + "_ID", None) or self.ID_PREFIX + "_" + operation
    )
    return self.pulp_ctx.call(
        operation_id,
        non_blocking=non_blocking,
        parameters=parameters,
        body=body,
        validate_body=validate_body,
    )

def preprocess_entity(self, body: EntityDefinition, partial: bool = False) -> EntityDefinition:
    """
    Filter to prepare the body for a create or update call.

    This function can be subclassed by specific Entity contexts to fix data depending on plugin
    versions.

    Parameters:
        body: The payload representing the entity to create or the fields to update on it.
        partial: Should be set if `body` may only represent part of the entity.

    Returns:
        The body ready to be passed to `call`.
    """
    return self.preprocess_body(body)

def list(self, limit: int, offset: int, parameters: t.Dict[str, t.Any]) -> t.List[t.Any]:
    """
    List entities by the type of this context.

    Parameters:
        limit: Maximal number of entities to return.
        offset: Number of entities to skip at the front of the list.
        parameters: Additional search or sorting criteria.

    Returns:
        List of entities matching the conditions.
    """
    count: int = -1
    entities: t.List[t.Any] = []
    payload: t.Dict[str, t.Any] = parameters.copy()
    payload.update(self.scope)
    payload["offset"] = offset
    payload["limit"] = BATCH_SIZE
    while limit != 0:
        if limit > BATCH_SIZE:
            limit -= BATCH_SIZE
        else:
            payload["limit"] = limit
            limit = 0
        result: t.Mapping[str, t.Any] = self.call("list", parameters=payload)
        count = result["count"]
        entities.extend(result["results"])
        if result["next"] is None:
            break
        payload["offset"] += payload["limit"]
    else:
        self.pulp_ctx.echo(
            _("Not all {count} entries were shown.").format(count=count), err=True
        )
    return entities

def find(self, **kwargs: t.Any) -> t.Any:
    """
    Find an entity based on search parameters.

    Note: It is preferred to use the `entity` property instead of calling `find` directly.

    Parameters:
        kwargs: The search parameters.

    Returns:
        The entity if one was found uniquely.

    Raises:
        PulpException: if no entity satisfies the search parameters uniquely.
    """
    payload: t.Dict[str, t.Any] = kwargs.copy()
    payload.update(self.scope)
    payload["offset"] = 0
    payload["limit"] = 1
    result: t.Mapping[str, t.Any] = self.call("list", parameters=payload)
    if result["count"] == 0:
        raise PulpException(
            _("Could not find {entity} with {kwargs}.").format(
                entity=self.ENTITY, kwargs=kwargs
            )
        )
    if result["count"] > 1:
        raise PulpException(
            _("Multiple {entities} found with {kwargs}.").format(
                entities=self.ENTITIES, kwargs=kwargs
            )
        )
    return result["results"][0]

def show(self, href: t.Optional[str] = None) -> t.Any:
    """
    Retrieve and return the full record of an entity from the server.

    Parameters:
        href: href of the entity to fetch. If not specified, the entity represented by `entity`
            will be used.

    Returns:
        The full record of the entity as reported by the server.

    Raises:
        PulpException: if no href was specified and the lazy lookup failed.
    """
    return self.call("read", parameters={self.HREF: href or self.pulp_href})

def create(
    self,
    body: EntityDefinition,
    parameters: t.Optional[t.Mapping[str, t.Any]] = None,
    non_blocking: bool = False,
) -> t.Any:
    """
    Create an entity.

    Parameters:
        body: Fields off the new entity and the values they should be set to.
        parameters: Additional parameters for the call (usually not needed).
        non_blocking: Whether the result of the operation should be awaited on.

    Returns:
        The created entity, or the record of the create task if non_blocking.
    """
    _parameters = self.scope
    if parameters:
        _parameters.update(parameters)
    if body is not None:
        body = self.preprocess_entity(body, partial=False)
    result = self.call(
        "create",
        parameters=_parameters,
        body=body,
        non_blocking=non_blocking,
    )
    if result["pulp_href"].startswith(self.pulp_ctx.api_path + "tasks/"):
        if not non_blocking:
            self.pulp_href = result["created_resources"][0]
            result = self.entity
    else:
        self._entity = result

    return result

def update(
    self,
    href: t.Optional[str] = None,
    body: t.Optional[EntityDefinition] = None,
    parameters: t.Optional[t.Mapping[str, t.Any]] = None,
    non_blocking: bool = False,
) -> t.Any:
    """
    Update the entity.

    Parameters:
        body: Fields and the values they should be changed to.
        parameters: Additional parameters for the call (usually not needed).
        non_blocking: Whether the result of the operation should be awaited on.
        href: **[DEPRECATED]**

    Returns:
        The updated entity, or the record of the update task if non_blocking.
    """
    _parameters = {self.HREF: href or self.pulp_href}
    if parameters:
        _parameters.update(parameters)
    if body is not None:
        body = self.preprocess_entity(body, partial=False)
    result = self.call(
        "partial_update",
        parameters=_parameters,
        body=body,
        non_blocking=non_blocking,
    )
    if result["pulp_href"].startswith(self.pulp_ctx.api_path + "tasks/"):
        if not non_blocking:
            self.pulp_href = self.pulp_href  # reload from server
            result = self.entity
    else:
        self._entity = result

    return result

def delete(self, href: t.Optional[str] = None, non_blocking: bool = False) -> t.Any:
    """
    Delete the entity.

    Parameters:
        non_blocking: Whether the result of the operation should be awaited on.
        href: **[DEPRECATED]**

    Returns:
        The record of the delete task.
    """
    return self.call(
        "delete", parameters={self.HREF: href or self.pulp_href}, non_blocking=non_blocking
    )

def set_label(self, key: str, value: str, non_blocking: bool = False) -> t.Any:
    """
    Set a label.

    Parameters:
        key: Name of the label.
        value: Value of the label.
        non_blocking: Whether the result of the operation should be awaited on.
            This no longer relevant for pulpcore>=3.34, because the synchronous api is used.
    """
    if self.pulp_ctx.has_plugin(PluginRequirement("core", specifier=">=3.34.0")):
        try:
            return self.call(
                "set_label",
                parameters={self.HREF: self.pulp_href},
                body={"key": key, "value": value},
            )
        except PulpHTTPError as e:
            if e.status_code != 403:
                raise
            # Workaround for broken access policies: Try the old mechanism.
    labels = self.entity["pulp_labels"]
    labels[key] = value
    return self.update(body={"pulp_labels": labels}, non_blocking=non_blocking)

def unset_label(self, key: str, non_blocking: bool = False) -> t.Any:
    """
    Unset a label.

    Parameters:
        key: Name of the label.
        non_blocking: Whether the result of the operation should be awaited on.
            This no longer relevant for pulpcore>=3.34, because the synchronous api is used.
    """
    if self.pulp_ctx.has_plugin(PluginRequirement("core", specifier=">=3.34.0")):
        try:
            return self.call(
                "unset_label", parameters={self.HREF: self.pulp_href}, body={"key": key}
            )
        except PulpHTTPError as e:
            if e.status_code != 403:
                raise
            # Workaround for broken access policies: Try the old mechanism.
    labels = self.entity["pulp_labels"]
    try:
        labels.pop(key)
    except KeyError:
        raise PulpException(_("Could not find label with key '{key}'.").format(key=key))
    return self.update(body={"pulp_labels": labels}, non_blocking=non_blocking)

def show_label(self, key: str) -> t.Optional[str]:
    """
    Show value of a label.

    Parameters:
        key: Name of the label.

    Returns:
        Value of the label or None.

    Raises:
        PulpException: if the label was not set.
    """
    labels: t.Dict[str, t.Optional[str]] = self.entity["pulp_labels"]
    try:
        return labels[key]
    except KeyError:
        raise PulpException(_("Could not find label with key '{key}'.").format(key=key))

def capable(self, capability: str) -> bool:
    """
    Report on a capability based on the presence of all needed server plugins.

    Parameters:
        capability: Name of a capability.

    Returns:
        Whether the capability is provided for this context.
    """
    return capability in self.CAPABILITIES and all(
        (self.pulp_ctx.has_plugin(pr) for pr in self.CAPABILITIES[capability])
    )

def needs_capability(self, capability: str) -> None:
    """
    Translates a capability in calls to `needs_plugin` via `CAPABILITIES`.

    Parameters:
        capability: Name of a capability.
    """
    if capability in self.CAPABILITIES:
        for plugin_requirement in self.CAPABILITIES[capability]:
            self.pulp_ctx.needs_plugin(plugin_requirement)
    else:
        raise PulpException(
            _("Capability '{capability}' needed on '{entity}' for this command.").format(
                capability=capability, entity=self.ENTITY
            )
        )

def repair(self, href: t.Optional[str] = None) -> t.Any:
    """
    Trigger a repair task for this repository version.

    Parameters:
        href: **[DEPRECATED]**

    Returns:
        The record of the repair task.
    """
    return self.call("repair", parameters={self.HREF: href or self.pulp_href}, body={})

def get_version_context(
    self,
    number: t.Optional[int] = None,
) -> PulpRepositoryVersionContext:
    """
    Return a repository version context of the proper type scoped for this repository.

    Parameters:
        number: Version number or `-1` for the latest version.

    Returns:
        Repository version context attached to the same pulp_ctx and scoped to the repository.
    """
    if number is not None:
        if number >= 0:
            version_href = self.entity["versions_href"] + f"{number}/"
        elif number == -1:
            version_href = self.entity["latest_version_href"]
        else:
            PulpException(_("Invalid version number ({number}).").format(number=number))
    else:
        version_href = None
    return self.VERSION_CONTEXT(
        pulp_ctx=self.pulp_ctx, repository_ctx=self, pulp_href=version_href
    )

def sync(
    self, href: t.Optional[str] = None, body: t.Optional[EntityDefinition] = None
) -> t.Any:
    """
    Trigger a sync task for this repository.

    Parameters:
        body: Any additional options specific to the repository type used to perform this sync.
        href: [DEPRECATED]

    Returns:
        Record of the sync task.
    """
    return self.call("sync", parameters={self.HREF: href or self.pulp_href}, body=body or {})

def modify(
    self,
    href: t.Optional[str] = None,
    add_content: t.Optional[t.List[str]] = None,
    remove_content: t.Optional[t.List[str]] = None,
    base_version: t.Optional[str] = None,
) -> t.Any:
    """
    Add to or remove content from this repository.

    Parameters:
        add_content: List of content hrefs to add.
        remove_content: List of content hrefs to remove.
        base_version: Href to a repository version relative to whose content the changes are to
            be interpreted.
        href: [DEPRECATED]

    Returns:
        Record of the modify task.
    """
    body: t.Dict[str, t.Any] = {}
    if add_content is not None:
        body["add_content_units"] = add_content
    if remove_content is not None:
        body["remove_content_units"] = remove_content
    if base_version is not None:
        body["base_version"] = base_version
    return self.call("modify", parameters={self.HREF: href or self.pulp_href}, body=body)

def reclaim(
    self,
    repo_hrefs: t.List[t.Union[str, PulpRepositoryContext]],
    repo_versions_keeplist: t.Optional[
        t.List[t.Union[str, PulpRepositoryVersionContext]]
    ] = None,
) -> t.Any:
    """
    Reclaim disk space for a list of repositories.

    Parameters:
        repo_hrefs: List of repository hrefs to reclaim.
        repo_versions_keeplist: List of repository version hrefs to keep unaffected.

    Returns:
        Record of the reclaim space task.
    """
    self.pulp_ctx.needs_plugin(PluginRequirement("core", specifier=">=3.19.0"))
    body: t.Dict[str, t.Any] = {}
    body["repo_hrefs"] = repo_hrefs
    if repo_versions_keeplist:
        body["repo_versions_keeplist"] = repo_versions_keeplist
    return self.call("reclaim_space_reclaim", body=body)

def upload(
    self,
    file: t.IO[bytes],
    chunk_size: int,
    repository: t.Optional[PulpRepositoryContext],
    **kwargs: t.Any,
) -> t.Any:
    """
    Create a content unit by uploading a file.

    Parameters:
        file: A file like object that supports `os.path.getsize`.
        chunk_size: Size of the chunks to upload independently.
        repository: Repository context to add the newly created content to.
        kwargs: Extra args specific to the content type, passed to the create call.

    Returns:
        The result of the create task.
    """
    self.needs_capability("upload")
    size = os.path.getsize(file.name)
    body: t.Dict[str, t.Any] = {**kwargs}
    if chunk_size > size:
        body["file"] = file
    elif self.pulp_ctx.has_plugin(PluginRequirement("core", specifier=">=3.20.0")):
        from pulp_glue.core.context import PulpUploadContext

        upload_href = PulpUploadContext(self.pulp_ctx).upload_file(file, chunk_size)
        body["upload"] = upload_href
    else:
        from pulp_glue.core.context import PulpArtifactContext

        artifact_href = PulpArtifactContext(self.pulp_ctx).upload(file, chunk_size)
        body["artifact"] = artifact_href
    if repository:
        body["repository"] = repository
    return self.create(body=body)