pulp_glue.common.context

PluginRequirement

PluginRequirement(
    name: str,
    min: Optional[str] = None,
    max: Optional[str] = None,
    feature: Optional[str] = None,
    inverted: bool = False,
    specifier: Optional[Union[str, SpecifierSet]] = None,
)

A class to represent a Pulp plugin with a set of versions.

This can be used in conjunction with has_plugin(s), needs_plugin(s) and CAPABILITIES. The parameters min and max are deprecated.

Parameters:
  • name (str) –

    The app-label of the pluin as reported by the status API.

  • feature (Optional[str], default: None ) –

    A string being displayed as the feature if used with needs_plugin and the condition is not met.

  • inverted (bool, default: False ) –

    Treat the version set in specifier as inverted. If no specifier is provided, this describes the requirement of a plugin not being installed.

  • specifier (Optional[Union[str, SpecifierSet]], default: None ) –

    A PEP-440 compatible version range.

  • min (Optional[str], default: None ) –

    [DEPRECATED]

  • max (Optional[str], default: None ) –

    [DEPRECATED]

Source code in pulp-glue/pulp_glue/common/context.py
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def __init__(
    self,
    name: str,
    min: Optional[str] = None,
    max: Optional[str] = None,
    feature: Optional[str] = None,
    inverted: bool = False,
    specifier: Optional[Union[str, SpecifierSet]] = None,
):
    self.name = name
    self.feature = feature
    self.inverted = inverted
    if min is None and max is None:
        specifier = specifier or ""
    else:
        assert specifier is None
        specifier = ""
        separator = ""
        if min:
            specifier += f">={min}"
            separator = ","
        if max:
            specifier += f"{separator}<{max}"
    if isinstance(specifier, SpecifierSet):
        self.specifier = specifier
    else:
        self.specifier = SpecifierSet(specifier, prereleases=True)

PulpException

Bases: Exception

The base exception pulp-glue will emit on expected error paths.

PulpHTTPError

PulpHTTPError(msg: str, status_code: int)

Bases: PulpException

Exception to indicate HTTP error responses.

Source code in pulp-glue/pulp_glue/common/context.py
111
112
113
def __init__(self, msg: str, status_code: int) -> None:
    super().__init__(msg)
    self.status_code = status_code

PulpNoWait

Bases: Exception

Exception to indicate that a task continues running in the background.

PulpContext

PulpContext(
    api_root: str,
    api_kwargs: Dict[str, Any],
    background_tasks: bool,
    timeout: Union[int, datetime.timedelta],
    domain: str = "default",
    format: Optional[str] = None,
)

Abstract class for the global PulpContext object. It is an abstraction layer for api access and output handling.

Parameters:
  • api_root (str) –

    The base url (excluding "api/v3/") to the servers 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 (Union[int, timedelta]) –

    Limit of time to wait for unfinished tasks.

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

    Name of the domain to interact with.

  • format (Optional[str], default: None ) –

    [DEPRECATED]

Source code in pulp-glue/pulp_glue/common/context.py
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
def __init__(
    self,
    api_root: str,
    api_kwargs: Dict[str, Any],
    background_tasks: bool,
    timeout: Union[int, datetime.timedelta],
    domain: str = "default",
    format: Optional[str] = None,  # Deprecated
) -> None:
    self._api: Optional[OpenAPI] = None
    self._api_root: str = api_root
    self._api_kwargs = api_kwargs
    self._needed_plugins: List[PluginRequirement] = [
        PluginRequirement("core", specifier=">=3.11.0")
    ]
    self.isatty: bool = sys.stdout.isatty()
    self.pulp_domain: str = domain

    self.background_tasks: bool = background_tasks
    if isinstance(timeout, int):
        timeout = datetime.timedelta(seconds=timeout)
    self.timeout: datetime.timedelta = timeout

api property

api: OpenAPI

The lazy evaluated OpenAPI object contained in this context.

This is only needed for low level interactions with the openapi spec. All calls to the API should be performed via call.

echo

echo(
    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.

Source code in pulp-glue/pulp_glue/common/context.py
242
243
244
245
246
247
248
249
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.")

prompt

prompt(text: str, hide_input: bool = False) -> Any

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.

Source code in pulp-glue/pulp_glue/common/context.py
251
252
253
254
255
256
257
258
def prompt(self, text: str, hide_input: bool = False) -> Any:
    """
    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.")

call

call(
    operation_id: str,
    non_blocking: bool = False,
    parameters: Optional[Dict[str, Any]] = None,
    body: Optional[EntityDefinition] = None,
    validate_body: bool = True,
) -> 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 (str) –

    The operation ID in the openapi v3 spec to be called.

  • non_blocking (bool, default: False ) –

    returns unfinished tasks if True.

  • parameters (Optional[Dict[str, Any]], default: None ) –

    Arguments that are to be sent as headers, querystrings or part of the URI.

  • body (Optional[EntityDefinition], default: None ) –

    Body payload for POST, PUT, PATCH calls.

  • validate_body (bool, default: True ) –

    Indicate whether the body should be validated.

Returns:
  • Any

    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.

Source code in pulp-glue/pulp_glue/common/context.py
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
def call(
    self,
    operation_id: str,
    non_blocking: bool = False,
    parameters: Optional[Dict[str, Any]] = None,
    body: Optional[EntityDefinition] = None,
    validate_body: bool = True,
) -> 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

wait_for_task

wait_for_task(
    task: EntityDefinition, expect_cancel: bool = False
) -> Any

Wait for a task to finish and return the finished task object.

Parameters:
  • task (EntityDefinition) –

    A task object to monitor.

  • expect_cancel (bool, default: False ) –

    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.

Source code in pulp-glue/pulp_glue/common/context.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
459
def wait_for_task(self, task: EntityDefinition, expect_cancel: bool = False) -> 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))

wait_for_task_group

wait_for_task_group(task_group: EntityDefinition) -> Any

Wait for a task group to finish and return the finished task object.

Parameters:
  • task_group (EntityDefinition) –

    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.

Source code in pulp-glue/pulp_glue/common/context.py
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
def wait_for_task_group(self, task_group: EntityDefinition) -> 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"]
            )
        )

PulpEntityContext

PulpEntityContext(
    pulp_ctx: PulpContext,
    pulp_href: Optional[str] = None,
    entity: Optional[EntityDefinition] = None,
)

Base class for entity specific contexts. This class provides the basic CRUD commands and ties its instances to the global PulpContext for api access. Mostly specification is achieved by defining / extending the class attributes below.

Parameters:
  • pulp_ctx (PulpContext) –

    The server context to attach this entity to.

  • pulp_href (Optional[str], default: None ) –

    Specifying this is equivalent to assinging to pulp_href later.

  • entity (Optional[EntityDefinition], default: None ) –

    Specifying this is equivalent to assinging to entity later.

Source code in pulp-glue/pulp_glue/common/context.py
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
def __init__(
    self,
    pulp_ctx: PulpContext,
    pulp_href: Optional[str] = None,
    entity: Optional[EntityDefinition] = None,
) -> None:
    assert pulp_href is None or entity is None

    self.meta: Dict[str, str] = {}
    self.pulp_ctx: PulpContext = pulp_ctx

    # Add requirements to the lazy evaluated list
    for plugin_requirement in self.NEEDS_PLUGINS:
        self.pulp_ctx.needs_plugin(plugin_requirement)

    self._entity = None
    if pulp_href is None:
        self._entity_lookup = entity or {}
    else:
        self.pulp_href = pulp_href

ENTITY class-attribute

ENTITY: str = _('entity')

Translatable name of the entity.

ENTITIES class-attribute

ENTITIES: str = _('entities')

Translatable plural of ENTITY.

HREF class-attribute

HREF: str

Name of the href parameter in the url patterns.

ID_PREFIX class-attribute

ID_PREFIX: str

Common prefix for the operations of this entity.

NULLABLES class-attribute

NULLABLES: Set[str] = set()

Set of fields that can be cleared by sending 'null'.

NEEDS_PLUGINS class-attribute

NEEDS_PLUGINS: List[PluginRequirement] = []

List of plugin requirements to operate such an entity on the server.

CAPABILITIES class-attribute

CAPABILITIES: Dict[str, List[PluginRequirement]] = {}

List of capabilities this entity provides.

Subclasses can specify version dependent capabilities here

Example
CAPABILITIES = {
    "feature1": [
        PluginRequirement("file"),
        PluginRequirement("core", specifier=">=3.7.0")
    ]
}

HREF_PATTERN instance-attribute

HREF_PATTERN: str

Regular expression with capture groups for 'plugin' and 'resource_type' to match URIs.

scope property

scope: Dict[str, Any]

Extra scope used in parameters for create and list calls.

Subclasses for nested entities can define the parameters for there parent scope here.

entity property writable

entity: EntityDefinition

Entity property that will perform a lazy lookup once it is accessed. You can specify lookup parameters by assigning a dictionary to it, or assign an href to the pulp_href property. To reset to having no attached entity you can assign None. Assigning to it will reset the lazy lookup behaviour.

pulp_href property writable

pulp_href: str

Property to represent the href of the attached entity. Assigning to it will reset the lazy lookup behaviour.

tangible property

tangible: bool

Indicate whether an entity is available or specified by search parameters.

call

call(
    operation: str,
    non_blocking: bool = False,
    parameters: Optional[Dict[str, Any]] = None,
    body: Optional[EntityDefinition] = None,
    validate_body: bool = True,
) -> 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 (str) –

    The operation to be performed on the entity. Usually the openapi operation_id is constructed by concatenating with the ID_PREFIX.

  • non_blocking (bool, default: False ) –

    returns unfinished tasks if True.

  • parameters (Optional[Dict[str, Any]], default: None ) –

    Arguments that are to be sent as headers, querystrings or part of the URI.

  • body (Optional[EntityDefinition], default: None ) –

    Body payload for POST, PUT, PATCH calls.

  • validate_body (bool, default: True ) –

    Indicate whether the body should be validated.

Returns:
  • Any

    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.

Source code in pulp-glue/pulp_glue/common/context.py
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
def call(
    self,
    operation: str,
    non_blocking: bool = False,
    parameters: Optional[Dict[str, Any]] = None,
    body: Optional[EntityDefinition] = None,
    validate_body: bool = True,
) -> 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,
    )

preprocess_entity

preprocess_entity(
    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 (EntityDefinition) –

    The payload representing the entity to create or the fields to update on it.

  • partial (bool, default: False ) –

    Should be set if body may only represent part of the entity.

Returns:
  • EntityDefinition

    The body ready to be passed to call.

Source code in pulp-glue/pulp_glue/common/context.py
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
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)

list

list(
    limit: int, offset: int, parameters: Dict[str, Any]
) -> List[Any]

List entities by the type of this context.

Parameters:
  • limit (int) –

    Maximal number of entities to return.

  • offset (int) –

    Number of entities to skip at the front of the list.

  • parameters (Dict[str, Any]) –

    Additional search or sorting criteria.

Returns:
  • List[Any]

    List of entities matching the conditions.

Source code in pulp-glue/pulp_glue/common/context.py
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
def list(self, limit: int, offset: int, parameters: Dict[str, Any]) -> List[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: List[Any] = []
    payload: Dict[str, 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: Mapping[str, 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

find

find(**kwargs: Any) -> Any

Find an entity based on search parameters.

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

Parameters:
  • kwargs (Any, default: {} ) –

    The search parameters.

Returns:
  • Any

    The entity if one was found uniquely.

Raises:
  • PulpException

    if no entity satisfies the search parameters uniquely.

Source code in pulp-glue/pulp_glue/common/context.py
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
def find(self, **kwargs: Any) -> 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: Dict[str, Any] = kwargs.copy()
    payload.update(self.scope)
    payload["offset"] = 0
    payload["limit"] = 1
    result: Mapping[str, 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]

show

show(href: Optional[str] = None) -> Any

Retrieve and return the full record of an entity from the server.

Parameters:
  • href (Optional[str], default: None ) –

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

Returns:
  • Any

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

Raises:
  • PulpException

    if no href was specified and the lazy lookup failed.

Source code in pulp-glue/pulp_glue/common/context.py
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
def show(self, href: Optional[str] = None) -> 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})

create

create(
    body: EntityDefinition,
    parameters: Optional[Mapping[str, Any]] = None,
    non_blocking: bool = False,
) -> Any

Create an entity.

Parameters:
  • body (EntityDefinition) –

    Fields off the new entity and the values they should be set to.

  • parameters (Optional[Mapping[str, Any]], default: None ) –

    Additional parameters for the call (usually not needed).

  • non_blocking (bool, default: False ) –

    Whether the result of the operation should be awaited on.

Returns:
  • Any

    The created entity, or the record of the create task if non_blocking.

Source code in pulp-glue/pulp_glue/common/context.py
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
def create(
    self,
    body: EntityDefinition,
    parameters: Optional[Mapping[str, Any]] = None,
    non_blocking: bool = False,
) -> 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

update

update(
    href: Optional[str] = None,
    body: Optional[EntityDefinition] = None,
    parameters: Optional[Mapping[str, Any]] = None,
    non_blocking: bool = False,
) -> Any

Update the entity.

Parameters:
  • body (Optional[EntityDefinition], default: None ) –

    Fields and the values they should be changed to.

  • parameters (Optional[Mapping[str, Any]], default: None ) –

    Additional parameters for the call (usually not needed).

  • non_blocking (bool, default: False ) –

    Whether the result of the operation should be awaited on.

  • href (Optional[str], default: None ) –

    [DEPRECATED]

Returns:
  • Any

    The updated entity, or the record of the update task if non_blocking.

Source code in pulp-glue/pulp_glue/common/context.py
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
def update(
    self,
    href: Optional[str] = None,
    body: Optional[EntityDefinition] = None,
    parameters: Optional[Mapping[str, Any]] = None,
    non_blocking: bool = False,
) -> 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

delete

delete(
    href: Optional[str] = None, non_blocking: bool = False
) -> Any

Delete the entity.

Parameters:
  • non_blocking (bool, default: False ) –

    Whether the result of the operation should be awaited on.

  • href (Optional[str], default: None ) –

    [DEPRECATED]

Returns:
  • Any

    The record of the delete task.

Source code in pulp-glue/pulp_glue/common/context.py
920
921
922
923
924
925
926
927
928
929
930
931
932
933
def delete(self, href: Optional[str] = None, non_blocking: bool = False) -> 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
    )

set_label

set_label(
    key: str, value: str, non_blocking: bool = False
) -> Any

Set a label.

Parameters:
  • key (str) –

    Name of the label.

  • value (str) –

    Value of the label.

  • non_blocking (bool, default: False ) –

    Whether the result of the operation should be awaited on. This no longer relevant for pulpcore>=3.34, because the synchronous api is used.

Source code in pulp-glue/pulp_glue/common/context.py
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
def set_label(self, key: str, value: str, non_blocking: bool = False) -> 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)

unset_label

unset_label(key: str, non_blocking: bool = False) -> Any

Unset a label.

Parameters:
  • key (str) –

    Name of the label.

  • non_blocking (bool, default: False ) –

    Whether the result of the operation should be awaited on. This no longer relevant for pulpcore>=3.34, because the synchronous api is used.

Source code in pulp-glue/pulp_glue/common/context.py
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
def unset_label(self, key: str, non_blocking: bool = False) -> 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)

show_label

show_label(key: str) -> t.Optional[str]

Show value of a label.

Parameters:
  • key (str) –

    Name of the label.

Returns:
  • Optional[str]

    Value of the label or None.

Raises:
Source code in pulp-glue/pulp_glue/common/context.py
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
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))

capable

capable(capability: str) -> bool

Report on a capability based on the presence of all needed server plugins.

Parameters:
  • capability (str) –

    Name of a capability.

Returns:
  • bool

    Whether the capability is provided for this context.

Source code in pulp-glue/pulp_glue/common/context.py
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
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])
    )

needs_capability

needs_capability(capability: str) -> None

Translates a capability in calls to needs_plugin via CAPABILITIES.

Parameters:
  • capability (str) –

    Name of a capability.

Source code in pulp-glue/pulp_glue/common/context.py
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
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
            )
        )

PulpRemoteContext

PulpRemoteContext(
    pulp_ctx: PulpContext,
    pulp_href: Optional[str] = None,
    entity: Optional[EntityDefinition] = None,
)

Bases: PulpEntityContext

Base class for remote contexts.

Source code in pulp-glue/pulp_glue/common/context.py
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
def __init__(
    self,
    pulp_ctx: PulpContext,
    pulp_href: Optional[str] = None,
    entity: Optional[EntityDefinition] = None,
) -> None:
    assert pulp_href is None or entity is None

    self.meta: Dict[str, str] = {}
    self.pulp_ctx: PulpContext = pulp_ctx

    # Add requirements to the lazy evaluated list
    for plugin_requirement in self.NEEDS_PLUGINS:
        self.pulp_ctx.needs_plugin(plugin_requirement)

    self._entity = None
    if pulp_href is None:
        self._entity_lookup = entity or {}
    else:
        self.pulp_href = pulp_href

PulpPublicationContext

PulpPublicationContext(
    pulp_ctx: PulpContext,
    pulp_href: Optional[str] = None,
    entity: Optional[EntityDefinition] = None,
)

Bases: PulpEntityContext

Base class for publication contexts.

Source code in pulp-glue/pulp_glue/common/context.py
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
def __init__(
    self,
    pulp_ctx: PulpContext,
    pulp_href: Optional[str] = None,
    entity: Optional[EntityDefinition] = None,
) -> None:
    assert pulp_href is None or entity is None

    self.meta: Dict[str, str] = {}
    self.pulp_ctx: PulpContext = pulp_ctx

    # Add requirements to the lazy evaluated list
    for plugin_requirement in self.NEEDS_PLUGINS:
        self.pulp_ctx.needs_plugin(plugin_requirement)

    self._entity = None
    if pulp_href is None:
        self._entity_lookup = entity or {}
    else:
        self.pulp_href = pulp_href

PulpDistributionContext

PulpDistributionContext(
    pulp_ctx: PulpContext,
    pulp_href: Optional[str] = None,
    entity: Optional[EntityDefinition] = None,
)

Bases: PulpEntityContext

Base class for distribution contexts.

Source code in pulp-glue/pulp_glue/common/context.py
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
def __init__(
    self,
    pulp_ctx: PulpContext,
    pulp_href: Optional[str] = None,
    entity: Optional[EntityDefinition] = None,
) -> None:
    assert pulp_href is None or entity is None

    self.meta: Dict[str, str] = {}
    self.pulp_ctx: PulpContext = pulp_ctx

    # Add requirements to the lazy evaluated list
    for plugin_requirement in self.NEEDS_PLUGINS:
        self.pulp_ctx.needs_plugin(plugin_requirement)

    self._entity = None
    if pulp_href is None:
        self._entity_lookup = entity or {}
    else:
        self.pulp_href = pulp_href

PulpRepositoryVersionContext

PulpRepositoryVersionContext(
    pulp_ctx: PulpContext,
    repository_ctx: PulpRepositoryContext,
    pulp_href: Optional[str] = None,
)

Bases: PulpEntityContext

Base class for repository version contexts.

Parameters:
  • pulp_ctx (PulpContext) –

    The server context to attach this entity to.

  • repository_ctx (PulpRepositoryContext) –

    Context of the repository this context should be scoped to.

  • pulp_href (Optional[str], default: None ) –

    Specifying this is equivalent to assinging to pulp_href later.

Source code in pulp-glue/pulp_glue/common/context.py
1147
1148
1149
1150
1151
1152
1153
1154
def __init__(
    self,
    pulp_ctx: PulpContext,
    repository_ctx: "PulpRepositoryContext",
    pulp_href: Optional[str] = None,
) -> None:
    super().__init__(pulp_ctx, pulp_href=pulp_href)
    self.repository_ctx = repository_ctx

repair

repair(href: Optional[str] = None) -> Any

Trigger a repair task for this repository version.

Parameters:
  • href (Optional[str], default: None ) –

    [DEPRECATED]

Returns:
  • Any

    The record of the repair task.

Source code in pulp-glue/pulp_glue/common/context.py
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
def repair(self, href: Optional[str] = None) -> 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={})

PulpRepositoryContext

PulpRepositoryContext(
    pulp_ctx: PulpContext,
    pulp_href: Optional[str] = None,
    entity: Optional[EntityDefinition] = None,
)

Bases: PulpEntityContext

Base class for repository contexts.

Source code in pulp-glue/pulp_glue/common/context.py
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
def __init__(
    self,
    pulp_ctx: PulpContext,
    pulp_href: Optional[str] = None,
    entity: Optional[EntityDefinition] = None,
) -> None:
    assert pulp_href is None or entity is None

    self.meta: Dict[str, str] = {}
    self.pulp_ctx: PulpContext = pulp_ctx

    # Add requirements to the lazy evaluated list
    for plugin_requirement in self.NEEDS_PLUGINS:
        self.pulp_ctx.needs_plugin(plugin_requirement)

    self._entity = None
    if pulp_href is None:
        self._entity_lookup = entity or {}
    else:
        self.pulp_href = pulp_href

get_version_context

get_version_context(
    number: Optional[int] = None,
) -> PulpRepositoryVersionContext

Return a repository version context of the proper type scoped for this repository.

Parameters:
  • number (Optional[int], default: None ) –

    Version number or -1 for the latest version.

Returns:
Source code in pulp-glue/pulp_glue/common/context.py
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
def get_version_context(
    self,
    number: 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
    )

sync

sync(
    href: Optional[str] = None,
    body: Optional[EntityDefinition] = None,
) -> Any

Trigger a sync task for this repository.

Parameters:
  • body (Optional[EntityDefinition], default: None ) –

    Any additional options specific to the repository type used to perform this sync.

  • href (Optional[str], default: None ) –

    [DEPRECATED]

Returns:
  • Any

    Record of the sync task.

Source code in pulp-glue/pulp_glue/common/context.py
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
def sync(self, href: Optional[str] = None, body: Optional[EntityDefinition] = None) -> 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 {})

modify

modify(
    href: Optional[str] = None,
    add_content: Optional[List[str]] = None,
    remove_content: Optional[List[str]] = None,
    base_version: Optional[str] = None,
) -> Any

Add to or remove content from this repository.

Parameters:
  • add_content (Optional[List[str]], default: None ) –

    List of content hrefs to add.

  • remove_content (Optional[List[str]], default: None ) –

    List of content hrefs to remove.

  • base_version (Optional[str], default: None ) –

    Href to a repository version relative to whose content the changes are to be interpreted.

  • href (Optional[str], default: None ) –

    [DEPRECATED]

Returns:
  • Any

    Record of the modify task.

Source code in pulp-glue/pulp_glue/common/context.py
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
def modify(
    self,
    href: Optional[str] = None,
    add_content: Optional[List[str]] = None,
    remove_content: Optional[List[str]] = None,
    base_version: Optional[str] = None,
) -> 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: Dict[str, 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)

PulpGenericRepositoryContext

PulpGenericRepositoryContext(
    pulp_ctx: PulpContext,
    pulp_href: Optional[str] = None,
    entity: Optional[EntityDefinition] = None,
)

Bases: PulpRepositoryContext

Generic repository context class to separate specific general functionality.

Source code in pulp-glue/pulp_glue/common/context.py
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
def __init__(
    self,
    pulp_ctx: PulpContext,
    pulp_href: Optional[str] = None,
    entity: Optional[EntityDefinition] = None,
) -> None:
    assert pulp_href is None or entity is None

    self.meta: Dict[str, str] = {}
    self.pulp_ctx: PulpContext = pulp_ctx

    # Add requirements to the lazy evaluated list
    for plugin_requirement in self.NEEDS_PLUGINS:
        self.pulp_ctx.needs_plugin(plugin_requirement)

    self._entity = None
    if pulp_href is None:
        self._entity_lookup = entity or {}
    else:
        self.pulp_href = pulp_href

reclaim

reclaim(
    repo_hrefs: List[t.Union[str, PulpRepositoryContext]],
    repo_versions_keeplist: Optional[
        List[t.Union[str, PulpRepositoryVersionContext]]
    ] = None,
) -> Any

Reclaim disk space for a list of repositories.

Parameters:
  • repo_hrefs (List[Union[str, PulpRepositoryContext]]) –

    List of repository hrefs to reclaim.

  • repo_versions_keeplist (Optional[List[Union[str, PulpRepositoryVersionContext]]], default: None ) –

    List of repository version hrefs to keep unaffected.

Returns:
  • Any

    Record of the reclaim space task.

Source code in pulp-glue/pulp_glue/common/context.py
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
def reclaim(
    self,
    repo_hrefs: List[t.Union[str, PulpRepositoryContext]],
    repo_versions_keeplist: Optional[List[t.Union[str, PulpRepositoryVersionContext]]] = None,
) -> 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: Dict[str, 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)

PulpContentContext

PulpContentContext(
    pulp_ctx: PulpContext,
    pulp_href: Optional[str] = None,
    entity: Optional[EntityDefinition] = None,
)

Bases: PulpEntityContext

Base class for content contexts.

Source code in pulp-glue/pulp_glue/common/context.py
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
def __init__(
    self,
    pulp_ctx: PulpContext,
    pulp_href: Optional[str] = None,
    entity: Optional[EntityDefinition] = None,
) -> None:
    assert pulp_href is None or entity is None

    self.meta: Dict[str, str] = {}
    self.pulp_ctx: PulpContext = pulp_ctx

    # Add requirements to the lazy evaluated list
    for plugin_requirement in self.NEEDS_PLUGINS:
        self.pulp_ctx.needs_plugin(plugin_requirement)

    self._entity = None
    if pulp_href is None:
        self._entity_lookup = entity or {}
    else:
        self.pulp_href = pulp_href

upload

upload(
    file: IO[bytes],
    chunk_size: int,
    repository: Optional[PulpRepositoryContext],
    **kwargs: Any
) -> Any

Create a content unit by uploading a file.

Parameters:
  • file (IO[bytes]) –

    A file like object that supports os.path.getsize.

  • chunk_size (int) –

    Size of the chunks to upload independently.

  • repository (Optional[PulpRepositoryContext]) –

    Repository context to add the newly created content to.

  • kwargs (Any, default: {} ) –

    Extra args specific to the content type, passed to the create call.

Returns:
  • Any

    The result of the create task.

Source code in pulp-glue/pulp_glue/common/context.py
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
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
def upload(
    self,
    file: IO[bytes],
    chunk_size: int,
    repository: Optional[PulpRepositoryContext],
    **kwargs: Any,
) -> 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: Dict[str, 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)

PulpACSContext

PulpACSContext(
    pulp_ctx: PulpContext,
    pulp_href: Optional[str] = None,
    entity: Optional[EntityDefinition] = None,
)

Bases: PulpEntityContext

Base class for ACS contexts.

Source code in pulp-glue/pulp_glue/common/context.py
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
def __init__(
    self,
    pulp_ctx: PulpContext,
    pulp_href: Optional[str] = None,
    entity: Optional[EntityDefinition] = None,
) -> None:
    assert pulp_href is None or entity is None

    self.meta: Dict[str, str] = {}
    self.pulp_ctx: PulpContext = pulp_ctx

    # Add requirements to the lazy evaluated list
    for plugin_requirement in self.NEEDS_PLUGINS:
        self.pulp_ctx.needs_plugin(plugin_requirement)

    self._entity = None
    if pulp_href is None:
        self._entity_lookup = entity or {}
    else:
        self.pulp_href = pulp_href

PulpContentGuardContext

PulpContentGuardContext(
    pulp_ctx: PulpContext,
    pulp_href: Optional[str] = None,
    entity: Optional[EntityDefinition] = None,
)

Bases: PulpEntityContext

Base class for content guard contexts.

Source code in pulp-glue/pulp_glue/common/context.py
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
def __init__(
    self,
    pulp_ctx: PulpContext,
    pulp_href: Optional[str] = None,
    entity: Optional[EntityDefinition] = None,
) -> None:
    assert pulp_href is None or entity is None

    self.meta: Dict[str, str] = {}
    self.pulp_ctx: PulpContext = pulp_ctx

    # Add requirements to the lazy evaluated list
    for plugin_requirement in self.NEEDS_PLUGINS:
        self.pulp_ctx.needs_plugin(plugin_requirement)

    self._entity = None
    if pulp_href is None:
        self._entity_lookup = entity or {}
    else:
        self.pulp_href = pulp_href

api_quirk

api_quirk(
    req: PluginRequirement,
) -> t.Callable[[t.Callable[[OpenAPI], None]], None]

A function decorator to allow manipulating API specs based on the availability of plugins.

Parameters:
  • req (PluginRequirement) –

    The plugin specifier to determine when the quirk should be applied.

Examples:

@api_quirk(PluginRequirement("catdog", specifier="<1.5.2"))
def patch_barking_filter_type(api: OpenAPI) -> None:
    # fixup api.api_spec here
Source code in pulp-glue/pulp_glue/common/context.py
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
def api_quirk(
    req: PluginRequirement,
) -> t.Callable[[t.Callable[[OpenAPI], None]], None]:
    """
    A function decorator to allow manipulating API specs based on the availability of plugins.

    Parameters:
        req: The plugin specifier to determine when the quirk should be applied.

    Examples:
        ```
        @api_quirk(PluginRequirement("catdog", specifier="<1.5.2"))
        def patch_barking_filter_type(api: OpenAPI) -> None:
            # fixup api.api_spec here
        ```
    """

    def _decorator(patch: t.Callable[[OpenAPI], None]) -> None:
        _REGISTERED_API_QUIRKS.append((req, patch))

    return _decorator