pulpcore.cli.common.generic

pass_pulp_context module-attribute

pass_pulp_context = make_pass_decorator(PulpCLIContext)

Decorator to make the Pulp context available to a command.

pass_entity_context module-attribute

pass_entity_context = make_pass_decorator(PulpEntityContext)

Decorator to make the nearest entity context available to a command.

pass_acs_context module-attribute

pass_acs_context = make_pass_decorator(PulpACSContext)

Decorator to make the nearest ACS context available to a command.

pass_content_context module-attribute

pass_content_context = make_pass_decorator(
    PulpContentContext
)

Decorator to make the nearest content context available to a command.

pass_repository_context module-attribute

pass_repository_context = make_pass_decorator(
    PulpRepositoryContext
)

Decorator to make the nearest repository context available to a command.

pass_repository_version_context module-attribute

pass_repository_version_context = make_pass_decorator(
    PulpRepositoryVersionContext
)

Decorator to make the nearest repository version context available to a command.

load_string_callback module-attribute

load_string_callback = load_file_wrapper(lambda , , : x)

A reusable callback for text parameters.

It will read data from a file if their value starts with "@", otherwise use it unchanged.

load_json_callback module-attribute

load_json_callback = load_file_wrapper(json_callback)

A reusable callback that will parse its value from json.

Will optionally read from a file prefixed with "@".

IncompatibleContext

Bases: UsageError

Exception to signal that an option or subcommand was used with an incompatible context.

ClickNoWait

Bases: ClickException

Exception raised when a user interrupts waiting for a task/taskgroup.

PulpCLIContext

PulpCLIContext(
    api_root: str,
    api_kwargs: t.Dict[str, t.Any],
    background_tasks: bool,
    timeout: int,
    format: str,
    domain: str = "default",
)

Bases: PulpContext

Subclass of the Context that overwrites the CLI specifics.

Parameters:
  • api_root (str) –

    The base url (excluding "api/v3/") to the server's api.

  • api_kwargs (Dict[str, Any]) –

    Extra arguments to pass to the wrapped OpenAPI object.

  • background_tasks (bool) –

    Whether to wait for tasks. If True, all tasks triggered will immediately raise PulpNoWait.

  • timeout (int) –

    Limit of time (in seconds) to wait for unfinished tasks.

  • format (str) –

    The format to be used by output_result.

  • domain (str, default: 'default' ) –

    Name of the domain to interact with.

Source code in pulpcore/cli/common/generic.py
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def __init__(
    self,
    api_root: str,
    api_kwargs: t.Dict[str, t.Any],
    background_tasks: bool,
    timeout: int,
    format: str,
    domain: str = "default",
) -> None:
    super().__init__(
        api_root=api_root,
        api_kwargs=api_kwargs,
        background_tasks=background_tasks,
        timeout=timeout,
        domain=domain,
    )
    self.format = format

output_result

output_result(result: t.Any) -> None

Dump the provided result to the console using the selected renderer.

Parameters:
  • result (Any) –

    JSON serializable data to be outputted.

Source code in pulpcore/cli/common/generic.py
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def output_result(self, result: t.Any) -> None:
    """
    Dump the provided result to the console using the selected renderer.

    arguments:
        result: JSON serializable data to be outputted.
    """
    if self.format == "json":
        output = json.dumps(result, cls=PulpJSONEncoder, indent=(2 if self.isatty else None))
        if PYGMENTS and self.isatty:
            output = highlight(output, JsonLexer(), Terminal256Formatter(style=PYGMENTS_STYLE))
        self.echo(output)
    elif self.format == "yaml":
        output = yaml.dump(result)
        if PYGMENTS and self.isatty:
            output = highlight(output, YamlLexer(), Terminal256Formatter(style=PYGMENTS_STYLE))
        self.echo(output)
    elif self.format == "none":
        pass
    else:
        raise NotImplementedError(
            _("Format '{format}' not implemented.").format(format=self.format)
        )

PulpOption

PulpOption(
    *args: t.Any,
    needs_plugins: t.Optional[
        t.List[PluginRequirement]
    ] = None,
    allowed_with_contexts: t.Optional[
        t.Tuple[t.Type[PulpEntityContext]]
    ] = None,
    **kwargs: t.Any
)

Bases: Option

Pulp-CLI specific subclass of click.Option.

The preferred way to use this is through the pulp_option factory.

Source code in pulpcore/cli/common/generic.py
306
307
308
309
310
311
312
313
314
315
def __init__(
    self,
    *args: t.Any,
    needs_plugins: t.Optional[t.List[PluginRequirement]] = None,
    allowed_with_contexts: t.Optional[t.Tuple[t.Type[PulpEntityContext]]] = None,
    **kwargs: t.Any,
):
    self.needs_plugins = needs_plugins
    self.allowed_with_contexts = allowed_with_contexts
    super().__init__(*args, **kwargs)

int_or_empty

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.

Source code in pulpcore/cli/common/generic.py
158
159
160
161
162
163
164
165
166
167
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)

float_or_empty

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.

Source code in pulpcore/cli/common/generic.py
173
174
175
176
177
178
179
180
181
182
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)

pulp_command

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.

Source code in pulpcore/cli/common/generic.py
273
274
275
276
277
278
279
280
281
282
283
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)

pulp_group

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.

Source code in pulpcore/cli/common/generic.py
286
287
288
289
290
291
292
293
294
295
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)

load_file_wrapper

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.

Source code in pulpcore/cli/common/generic.py
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
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

json_callback

json_callback(
    ctx: click.Context,
    param: click.Parameter,
    value: t.Optional[str],
) -> t.Any

A reusable callback that will parse its value from json.

Source code in pulpcore/cli/common/generic.py
469
470
471
472
473
474
475
476
477
478
479
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

pulp_option

pulp_option(
    *args: t.Any, **kwargs: t.Any
) -> t.Callable[[FC], FC]

Factory of 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,),
)

Source code in pulpcore/cli/common/generic.py
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
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)

list_command

list_command(**kwargs: t.Any) -> click.Command

A factory that creates a list command.

Source code in pulpcore/cli/common/generic.py
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
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

show_command

show_command(**kwargs: t.Any) -> click.Command

A factory that creates a show command.

Source code in pulpcore/cli/common/generic.py
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
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

create_command

create_command(**kwargs: t.Any) -> click.Command

A factory that creates a create command.

Source code in pulpcore/cli/common/generic.py
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
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

update_command

update_command(**kwargs: t.Any) -> click.Command

A factory that creates an update command.

Source code in pulpcore/cli/common/generic.py
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
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

destroy_command

destroy_command(**kwargs: t.Any) -> click.Command

A factory that creates a destroy command.

Source code in pulpcore/cli/common/generic.py
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
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

version_command

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.

Source code in pulpcore/cli/common/generic.py
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
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

label_command

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.

Source code in pulpcore/cli/common/generic.py
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
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

role_command

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.

Source code in pulpcore/cli/common/generic.py
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
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

repository_content_command

repository_content_command(**kwargs: t.Any) -> click.Group

A factory that creates a repository content command group.

Source code in pulpcore/cli/common/generic.py
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
1530
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
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