pulpcore.cli.common.generic¶

def int_or_empty(value: str) -> t.Union[str, int]:
    """
    Turns a string into an integer or keeps the empty string.

    This is meant to be used as a click parameter type.
    """
    if value == "":
        return ""
    else:
        return int(value)

def float_or_empty(value: str) -> t.Union[str, float]:
    """
    Turns a string into a float or keeps the empty string.

    This is meant to be used as a click parameter type.
    """
    if value == "":
        return ""
    else:
        return float(value)

def pulp_command(
    name: t.Optional[str] = None, **kwargs: t.Any
) -> t.Callable[[_AnyCallable], PulpCommand]:
    """
    Pulp command factory.

    Creates a click compatible command that can be modified with `needs_plugins` and
    `allowed_with_contexts`. It allows rendering the docstring with the values of `ENTITY` and
    `ENTITIES` from the closest entity context.
    """
    return click.command(name=name, cls=PulpCommand, **kwargs)

def pulp_group(
    name: t.Optional[str] = None, **kwargs: t.Any
) -> t.Callable[[_AnyCallable], PulpGroup]:
    """
    Pulp command group factory.

    Creates a click compatible group command that selects subcommands based on
    `allowed_with_contexts` and creates `PulpCommand` subcommands by default.
    """
    return click.group(name=name, cls=PulpGroup, **kwargs)

def load_file_wrapper(handler: t.Callable[[click.Context, click.Parameter, str], t.Any]) -> t.Any:
    """
    A wrapper that is used for chaining or decorating callbacks that manipulate input data.

    When prefixed with `"@"`, content will be read from a file instead of being taken from the
    command line.
    """

    @wraps(handler)
    def _load_file_or_string_wrapper(
        ctx: click.Context, param: click.Parameter, value: t.Optional[str]
    ) -> t.Any:
        """Load the string from input, or from file if the value starts with @."""
        if not value:
            return value

        if value.startswith("@"):
            the_file = value[1:]
            try:
                with click.open_file(the_file, "r") as fp:
                    the_content = fp.read()
            except OSError:
                raise click.ClickException(
                    _("Failed to load content from {file}").format(file=the_file)
                )
        else:
            the_content = value

        return handler(ctx, param, the_content)

    return _load_file_or_string_wrapper

def json_callback(ctx: click.Context, param: click.Parameter, value: t.Optional[str]) -> t.Any:
    """A reusable callback that will parse its value from json."""
    if value is None:
        return None

    try:
        json_object = json.loads(value)
    except json.decoder.JSONDecodeError:
        raise click.ClickException(_("Failed to decode JSON"))
    else:
        return json_object

def pulp_option(*args: t.Any, **kwargs: t.Any) -> t.Callable[[FC], FC]:
    """
    Factory of [`PulpOption`][pulpcore.cli.common.generic.PulpOption] objects.

    `PulpOption` provides extra features over `click.Option`, namely:

    1. Define version constrains.
    2. Support for template variables in the help message.
    3. Limit the use of options to certain entity contexts.

    Examples:
        Define version constrains and custom help message:
        ```
        pulp_option(
            "--name",
            needs_plugins=[PluginRequirement("rpm", specifier=">=3.12.0")],
            help=_("Name of {entity}"),
            allowed_with_contexts=(PulpRpmRepositoryContext,),
        )
        ```
    """
    kwargs["cls"] = PulpOption
    return click.option(*args, **kwargs)

def list_command(**kwargs: t.Any) -> click.Command:
    """A factory that creates a list command."""

    kwargs.setdefault("name", "list")
    kwargs.setdefault("help", _("Show the list of optionally filtered {entities}."))
    decorators = kwargs.pop("decorators", [])

    # This is a mypy bug getting confused with positional args
    # https://github.com/python/mypy/issues/15037
    @pulp_command(**kwargs)  # type: ignore [arg-type]
    @limit_option
    @offset_option
    @ordering_option
    @field_option
    @exclude_field_option
    @pass_entity_context
    @pass_pulp_context
    def callback(
        pulp_ctx: PulpCLIContext,
        entity_ctx: PulpEntityContext,
        limit: int,
        offset: int,
        **kwargs: t.Any,
    ) -> None:
        """
        Show the list of optionally filtered {entities}.
        """
        if "ordering" in kwargs:
            # Workaround for missing ordering filter
            if not kwargs["ordering"]:
                kwargs["ordering"] = None
        result = entity_ctx.list(limit=limit, offset=offset, parameters=kwargs)
        pulp_ctx.output_result(result)

    for option in decorators:
        # Decorate callback
        callback = option(callback)
    return callback

def show_command(**kwargs: t.Any) -> click.Command:
    """A factory that creates a show command."""

    if "name" not in kwargs:
        kwargs["name"] = "show"
    if "help" not in kwargs:
        kwargs["help"] = _("Show details of a {entity}.")
    decorators = kwargs.pop("decorators", [])

    @pulp_command(**kwargs)
    @pass_entity_context
    @pass_pulp_context
    def callback(pulp_ctx: PulpCLIContext, entity_ctx: PulpEntityContext) -> None:
        """
        Show details of a {entity}.
        """
        pulp_ctx.output_result(entity_ctx.entity)

    for option in decorators:
        # Decorate callback
        callback = option(callback)
    return callback

def create_command(**kwargs: t.Any) -> click.Command:
    """A factory that creates a create command."""

    if "name" not in kwargs:
        kwargs["name"] = "create"
    if "help" not in kwargs:
        kwargs["help"] = _("Create a {entity}.")
    decorators = kwargs.pop("decorators", [])

    # This is a mypy bug getting confused with positional args
    # https://github.com/python/mypy/issues/15037
    @pulp_command(**kwargs)  # type: ignore [arg-type]
    @pass_entity_context
    @pass_pulp_context
    def callback(pulp_ctx: PulpCLIContext, entity_ctx: PulpEntityContext, **kwargs: t.Any) -> None:
        """
        Create a {entity}.
        """
        result = entity_ctx.create(body=kwargs)
        if "created_resources" in result:
            entity_ctx.pulp_href = result["created_resources"][0]
            result = entity_ctx.entity
        pulp_ctx.output_result(result)

    for option in decorators:
        # Decorate callback
        callback = option(callback)
    return callback

def update_command(**kwargs: t.Any) -> click.Command:
    """A factory that creates an update command."""

    if "name" not in kwargs:
        kwargs["name"] = "update"
    if "help" not in kwargs:
        kwargs["help"] = _("Update a {entity}.")
    decorators = kwargs.pop("decorators", [])

    # This is a mypy bug getting confused with positional args
    # https://github.com/python/mypy/issues/15037
    @pulp_command(**kwargs)  # type: ignore [arg-type]
    @pass_entity_context
    @pass_pulp_context
    def callback(pulp_ctx: PulpCLIContext, entity_ctx: PulpEntityContext, **kwargs: t.Any) -> None:
        """
        Update a {entity}.
        """
        entity_ctx.update(body=kwargs)

    for option in decorators:
        # Decorate callback
        callback = option(callback)
    return callback

def destroy_command(**kwargs: t.Any) -> click.Command:
    """A factory that creates a destroy command."""

    kwargs.setdefault("name", "destroy")
    kwargs.setdefault("help", _("Destroy a {entity}."))
    decorators = kwargs.pop("decorators", [])

    @pulp_command(**kwargs)
    @pass_entity_context
    def callback(entity_ctx: PulpEntityContext) -> None:
        """
        Destroy a {entity}.
        """
        entity_ctx.delete()

    for option in decorators:
        # Decorate callback
        callback = option(callback)
    return callback

def version_command(**kwargs: t.Any) -> click.Command:
    """
    A factory that creates a repository version command group.

    This group contains `list`, `show`, `destroy` and `repair` subcommands.
    If `list_only=True` is passed, only the `list` command will be instantiated.
    Repository lookup options can be provided in `decorators`.
    """

    kwargs.setdefault("name", "version")
    decorators = kwargs.pop("decorators", [repository_href_option, repository_option])
    list_only = kwargs.pop("list_only", False)

    @pulp_group(**kwargs)
    @pass_repository_context
    @click.pass_context
    def callback(
        ctx: click.Context,
        repository_ctx: PulpRepositoryContext,
    ) -> None:
        ctx.obj = repository_ctx.get_version_context()

    callback.add_command(list_command(decorators=decorators + [content_in_option]))

    if not list_only:
        callback.add_command(show_command(decorators=decorators + [version_option]))
        callback.add_command(destroy_command(decorators=decorators + [version_option]))

        @callback.command()
        @repository_option
        @version_option
        @pass_repository_version_context
        @pass_pulp_context
        def repair(
            pulp_ctx: PulpCLIContext,
            repository_version_ctx: PulpRepositoryVersionContext,
        ) -> None:
            result = repository_version_ctx.repair()
            pulp_ctx.output_result(result)

    return callback

def label_command(**kwargs: t.Any) -> click.Command:
    """
    A factory that creates a label command group.

    This group contains `set`, `unset` and `show` commands and acts on the nearest entity context.
    Pass options in as `decorators` to customize the entity lookup options.
    """

    kwargs.setdefault("name", "label")
    decorators = kwargs.pop("decorators", [name_option, href_option])
    need_plugins = kwargs.pop("need_plugins", [])

    @pulp_group(**kwargs)
    @pass_pulp_context
    def label_group(pulp_ctx: PulpCLIContext) -> None:
        for item in need_plugins:
            pulp_ctx.needs_plugin(item)

    @pulp_command(name="set", help=_("Add or update a label"))
    @click.option("--key", required=True, help=_("Key of the label"))
    @click.option("--value", required=True, help=_("Value of the label"))
    @pass_entity_context
    def label_set(entity_ctx: PulpEntityContext, key: str, value: str) -> None:
        """Add or update a label"""
        entity_ctx.set_label(key, value)

    @pulp_command(name="unset", help=_("Remove a label with a given key"))
    @click.option("--key", required=True, help=_("Key of the label"))
    @pass_entity_context
    def label_unset(entity_ctx: PulpEntityContext, key: str) -> None:
        """Remove a label with a given key"""
        entity_ctx.unset_label(key)

    @pulp_command(name="show", help=_("Show the value for a particular label key"))
    @click.option("--key", required=True, help=_("Key of the label"))
    @pass_entity_context
    def label_show(entity_ctx: PulpEntityContext, key: str) -> None:
        """Show the value for a particular label key"""
        click.echo(entity_ctx.show_label(key))

    for subcmd in [label_set, label_unset, label_show]:
        for decorator in decorators:
            subcmd = decorator(subcmd)
        label_group.add_command(subcmd)

    return label_group

def role_command(**kwargs: t.Any) -> click.Command:
    """
    A factory that creates a (object) role command group.

    This group contains `my-permissions`, `list`, `add` and `remove`.
    Pass options in as `decorators` to customize the entity lookup options.
    """

    kwargs.setdefault("name", "role")
    kwargs.setdefault("help", _("Manage object roles."))
    decorators = kwargs.pop("decorators", [name_option, href_option])

    @pulp_group(**kwargs)
    def role_group() -> None:
        pass

    @pulp_command(help=_("List my permissions on this object."))
    @pass_entity_context
    @pass_pulp_context
    def my_permissions(pulp_ctx: PulpCLIContext, entity_ctx: PulpEntityContext) -> None:
        result = entity_ctx.my_permissions()
        pulp_ctx.output_result(result)

    @pulp_command(name="list", help=_("List assigned object roles."))
    @pass_entity_context
    @pass_pulp_context
    def role_list(pulp_ctx: PulpCLIContext, entity_ctx: PulpEntityContext) -> None:
        result = entity_ctx.list_roles()
        pulp_ctx.output_result(result)

    @pulp_command(name="add", help=_("Add assigned object roles."))
    @click.option("--role")
    @click.option("--user", "users", multiple=True)
    @click.option("--group", "groups", multiple=True)
    @pass_entity_context
    @pass_pulp_context
    def role_add(
        pulp_ctx: PulpCLIContext,
        entity_ctx: PulpEntityContext,
        role: str,
        users: t.List[str],
        groups: t.List[str],
    ) -> None:
        result = entity_ctx.add_role(role, users, groups)
        pulp_ctx.output_result(result)

    @pulp_command(name="remove", help=_("Remove assigned object roles."))
    @click.option("--role")
    @click.option("--user", "users", multiple=True)
    @click.option("--group", "groups", multiple=True)
    @pass_entity_context
    @pass_pulp_context
    def role_remove(
        pulp_ctx: PulpCLIContext,
        entity_ctx: PulpEntityContext,
        role: str,
        users: t.List[str],
        groups: t.List[str],
    ) -> None:
        result = entity_ctx.remove_role(role, users, groups)
        pulp_ctx.output_result(result)

    for subcmd in [my_permissions, role_list, role_add, role_remove]:
        for decorator in decorators:
            subcmd = decorator(subcmd)
        role_group.add_command(subcmd)

    return role_group

def repository_content_command(**kwargs: t.Any) -> click.Group:
    """A factory that creates a repository content command group."""

    content_contexts = kwargs.pop("contexts", {})

    def version_callback(
        ctx: click.Context, param: click.Parameter, value: t.Optional[int]
    ) -> PulpRepositoryVersionContext:
        repo_ctx = ctx.find_object(PulpRepositoryContext)
        assert repo_ctx is not None
        return repo_ctx.get_version_context(-1 if value is None else value)

    # This is a mypy bug getting confused with positional args
    # https://github.com/python/mypy/issues/15037
    @pulp_command("list")  # type: ignore [arg-type]
    @click.option("--all-types", is_flag=True)
    @limit_option
    @offset_option
    @repository_option
    @click.option("--version", type=int, callback=version_callback)
    @pass_pulp_context
    @pass_content_context
    def content_list(
        content_ctx: PulpContentContext,
        pulp_ctx: PulpCLIContext,
        version: PulpRepositoryVersionContext,
        offset: int,
        limit: int,
        all_types: bool,
        **params: t.Any,
    ) -> None:
        parameters = {k: v for k, v in params.items() if v is not None}
        parameters.update({"repository_version": version.pulp_href})
        content_ctx = PulpContentContext(pulp_ctx) if all_types else content_ctx
        result = content_ctx.list(limit=limit, offset=offset, parameters=parameters)
        pulp_ctx.output_result(result)

    @pulp_command("add")
    @repository_option
    @click.option("--base-version", type=int, callback=version_callback)
    @pass_content_context
    def content_add(
        content_ctx: PulpContentContext,
        base_version: PulpRepositoryVersionContext,
    ) -> None:
        repo_ctx = base_version.repository_ctx
        repo_ctx.modify(add_content=[content_ctx.pulp_href], base_version=base_version.pulp_href)

    @pulp_command("remove")
    @click.option("--all", is_flag=True, help=_("Remove all content from repository version"))
    @repository_option
    @click.option("--base-version", type=int, callback=version_callback)
    @pass_content_context
    def content_remove(
        content_ctx: PulpContentContext,
        base_version: PulpRepositoryVersionContext,
        all: bool,
    ) -> None:
        repo_ctx = base_version.repository_ctx
        remove_content = ["*" if all else content_ctx.pulp_href]
        repo_ctx.modify(remove_content=remove_content, base_version=base_version.pulp_href)

    @pulp_command("modify")
    @repository_option
    @click.option("--base-version", type=int, callback=version_callback)
    def content_modify(
        base_version: PulpRepositoryVersionContext,
        add_content: t.Optional[t.List[PulpContentContext]],
        remove_content: t.Optional[t.List[PulpContentContext]],
    ) -> None:
        repo_ctx = base_version.repository_ctx
        ac = [unit.pulp_href for unit in add_content] if add_content else None
        rc = [unit.pulp_href for unit in remove_content] if remove_content else None
        repo_ctx.modify(add_content=ac, remove_content=rc, base_version=base_version.pulp_href)

    command_decorators: t.Dict[click.Command, t.Optional[t.List[t.Callable[[FC], FC]]]] = {
        content_list: kwargs.pop("list_decorators", []),
        content_add: kwargs.pop("add_decorators", None),
        content_remove: kwargs.pop("remove_decorators", None),
        content_modify: kwargs.pop("modify_decorators", None),
    }
    kwargs.setdefault("name", "content")

    @pulp_group(**kwargs)
    @type_option(choices=content_contexts)
    def content_group() -> None:
        pass

    for command, options in command_decorators.items():
        if options is not None:
            for option in options:
                command = option(command)
            content_group.add_command(command)

    return content_group