pulp_glue.common.context¶

`pulp_glue.common.context` ¶

`PluginRequirement(name, feature=None, inverted=False, specifier=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.

Parameters:

Name	Type	Description	Default
`name`	`str`	The app-label of the pluin as reported by the status API.	required
`feature`	`Optional[str]`	A string being displayed as the feature if used with `needs_plugin` and the condition is not met.	`None`
`inverted`	`bool`	Treat the version set in specifier as inverted. If no specifier is provided, this describes the requirement of a plugin not being installed.	`False`
`specifier`	`Optional[Union[str, SpecifierSet]]`	A PEP-440 compatible version range.	`None`

`PulpException` ¶

Bases: Exception

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

`PulpEntityNotFound` ¶

Bases: PulpException

Exception to signify that an entity was not found.

`PulpHTTPError(msg, status_code)` ¶

Bases: PulpException

Exception to indicate HTTP error responses.

`PulpNoWait` ¶

Bases: Exception

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

`PulpContext(api_root, api_kwargs, background_tasks=False, timeout=300, domain='default', fake_mode=False)` ¶

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

Parameters:

Name	Type	Description	Default
`api_root`	`str`	The base url (excluding "api/v3/") to the servers api.	required
`api_kwargs`	`Dict[str, Any]`	Extra arguments to pass to the wrapped `OpenAPI` object.	required
`background_tasks`	`bool`	Whether to wait for tasks. If `True`, all tasks triggered will immediately raise `PulpNoWait`.	`False`
`timeout`	`Union[int, timedelta]`	Limit of time to wait for unfinished tasks.	`300`
`domain`	`str`	Name of the domain to interact with.	`'default'`
`fake_mode`	`bool`	In fake mode, no modifying calls will be performed. Where possible, instead of failing, the requested result will be faked. This implies `safe_calls_only=True` on the `api_kwargs`.	`False`

`api: OpenAPI` `property` ¶

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(message, nl=True, err=False)` ¶

Abstract function that will be called to emit warnings and task progress.

Warning

This function does nothing until implemented by a subclass.

`prompt(text, hide_input=False)` ¶

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. Doing so is deprecated.

`from_config_files(profile=None, config_locations=None)` `classmethod` ¶

Create a PulpContext object from config files.

Note

This feature needs Python >=3.11 to work for now, because we don't want to add another dependency to pulp-glue.

Parameters:

Name	Type	Description	Default
`profile`	`Optional[str]`	Select a different profile from the config.	`None`
`config_locations`	`Optional[List[str]]`	If provided these config files will be merged (last on wins) instead of the default locations.	`None`

Returns: A configured PulpContext object.

`from_config(config)` `classmethod` ¶

Create a PulpContext object from a config dictionary.

Parameters:

Name	Type	Description	Default
`config`	`Dict[str, Any]`	dictionary of configuration values.	required

Returns: A configured PulpContext object.

`call(operation_id, non_blocking=False, parameters=None, body=None, validate_body=True)` ¶

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:

Name	Type	Description	Default
`operation_id`	`str`	The operation ID in the openapi v3 spec to be called.	required
`non_blocking`	`bool`	returns unfinished tasks if `True`.	`False`
`parameters`	`Optional[Dict[str, Any]]`	Arguments that are to be sent as headers, querystrings or part of the URI.	`None`
`body`	`Optional[EntityDefinition]`	Body payload for POST, PUT, PATCH calls.	`None`
`validate_body`	`bool`	Indicate whether the body should be validated.	`True`

Returns:

Type	Description
`Any`	The body of the response, or the task or task group if one was issued.

Raises:

Type	Description
`PulpNoWait`	in case the context has `background_tasks` set or a task (group) timed out.
`NotImplementedFake`	if an unsafe call was attempted in `fake_mode`
`PulpHTTPError`	for unhandeld REST API errors
`PulpException`	for all unhandeld openapi and http connection exceptions

`wait_for_task(task, expect_cancel=False)` ¶

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

Parameters:

Name	Type	Description	Default
`task`	`EntityDefinition`	A task object to monitor.	required
`expect_cancel`	`bool`	Swaps the raising condition for completed and canceled tasks.	`False`

Raises:

Type	Description
`PulpNoWait`	on timeout or if the context has `background_tasks` set.
`PulpException`	on ctrl-c, if task failed or was canceled.

`wait_for_task_group(task_group)` ¶

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

Parameters:

Name	Type	Description	Default
`task_group`	`EntityDefinition`	A task group object all of which's tasks to monitor.	required

Raises:

Type	Description
`PulpNoWait`	on timeout or if the context has `background_tasks` set.
`PulpException`	on ctrl-c, if task failed or was canceled.

`PulpViewSetContext(pulp_ctx)` ¶

Base class to interact with a generic viewset.

Parameters:

Name	Type	Description	Default
`pulp_ctx`	`PulpContext`	The server context to attach this viewset context to.	required

`ID_PREFIX: str` `class-attribute` ¶

Common prefix for the operations of this entity.

`NEEDS_PLUGINS: t.List[PluginRequirement] = []` `class-attribute` ¶

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

`call(operation, non_blocking=False, parameters=None, body=None, validate_body=True)` ¶

Perform an API call for operation. Wait for triggered tasks to finish if not background. Returns the operation result, or the finished task.

Parameters:

Name	Type	Description	Default
`operation`	`str`	The operation to be performed on the entity. Usually the openapi operation_id is constructed by concatenating with the `ID_PREFIX`.	required
`non_blocking`	`bool`	returns unfinished tasks if `True`.	`False`
`parameters`	`Optional[Dict[str, Any]]`	Arguments that are to be sent as headers, querystrings or part of the URI.	`None`
`body`	`Optional[EntityDefinition]`	Body payload for POST, PUT, PATCH calls.	`None`
`validate_body`	`bool`	Indicate whether the body should be validated.	`True`

Returns:

Type	Description
`Any`	The body of the response, or the task or task group if one was issued.

Raises:

Type	Description
`PulpNoWait`	in case the context has `background_tasks` set or a task (group) timed out.

`PulpEntityContext(pulp_ctx, pulp_href=None, entity=None)` ¶

Bases: PulpViewSetContext

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

Parameters:

Name	Type	Description	Default
`pulp_ctx`	`PulpContext`	The server context to attach this entity context to.	required
`pulp_href`	`Optional[str]`	Specifying this is equivalent to assinging to `pulp_href` later.	`None`
`entity`	`Optional[EntityDefinition]`	Specifying this is equivalent to assinging to `entity` later.	`None`

`ENTITY: str = _('entity')` `class-attribute` ¶

Translatable name of the entity.

`ENTITIES: str = _('entities')` `class-attribute` ¶

Translatable plural of ENTITY.

`HREF: str` `class-attribute` ¶

Name of the href parameter in the url patterns.

`NULLABLES: t.Set[str] = set()` `class-attribute` ¶

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

`CAPABILITIES: t.Dict[str, t.List[PluginRequirement]] = {}` `class-attribute` ¶

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: str` `instance-attribute` ¶

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

`scope: t.Dict[str, t.Any]` `property` ¶

Extra scope used in parameters for create and list calls.

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

`entity: EntityDefinition` `property` `writable` ¶

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.

!!!note: A # type: ignore[assignment] comment is needed due to a mypy limitation. https://github.com/python/mypy/issues/3004

Assigning to it will reset the lazy lookup behaviour.

`pulp_href: str` `property` `writable` ¶

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

`tangible: bool` `property` ¶

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

`preprocess_entity(body, partial=False)` ¶

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:

Name	Type	Description	Default
`body`	`EntityDefinition`	The payload representing the entity to create or the fields to update on it.	required
`partial`	`bool`	Should be set if `body` may only represent part of the entity.	`False`

Returns:

Type	Description
`EntityDefinition`	The body ready to be passed to `call`.

`list_iterator(parameters=None, offset=0, batch_size=BATCH_SIZE, stats=None)` ¶

List entities from this context in a batched iterator.

Parameters:

Name	Type	Description	Default
`parameters`	`Optional[Dict[str, Any]]`	Search or sorting criteria.	`None`
`offset`	`int`	Number of entities to skip.	`0`
`batch_size`	`int`	Size of the batches to fetch. Maximally BATCH_SIZE will be used.	`BATCH_SIZE`
`stats`	`Optional[Dict[str, Any]]`	If provided, a dictionary that will be filled with metadata: count: Number of entities reported by the server to match the criteria.	`None`

Returns:

Type	Description
`Iterator[Any]`	Iterator of entities matching the search conditions.

`list(limit, offset, parameters)` ¶

List entities by the type of this context.

Parameters:

Name	Type	Description	Default
`limit`	`int`	Maximal number of entities to return Use 0 to loop until all entries are retrieved.	required
`offset`	`int`	Number of entities to skip at the front of the list.	required
`parameters`	`Dict[str, Any]`	Additional search or sorting criteria.	required

Returns:

Type	Description
`List[Any]`	List of entities matching the conditions.

`find(**kwargs)` ¶

Find an entity based on search parameters.

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

Parameters:

Name	Type	Description	Default
`kwargs`	`Any`	The search parameters.	`{}`

Returns:

Type	Description
`Any`	The entity if one was found uniquely.

Raises:

Type	Description
`PulpEntityNotFound`	if no entity satisfies the search parameters.
`PulpException`	if multiple entities satisfy the search parameters.

`show(href=None)` ¶

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

Parameters:

Name	Type	Description	Default
`href`	`Optional[str]`	href of the entity to fetch. If not specified, the entity represented by `entity` will be used.	`None`

Returns:

Type	Description
`Any`	The full record of the entity as reported by the server.

Raises:

Type	Description
`PulpException`	if no href was specified and the lazy lookup failed.

`create(body, parameters=None, non_blocking=False)` ¶

Create an entity.

Parameters:

Name	Type	Description	Default
`body`	`EntityDefinition`	Fields off the new entity and the values they should be set to.	required
`parameters`	`Optional[Mapping[str, Any]]`	Additional parameters for the call (usually not needed).	`None`
`non_blocking`	`bool`	Whether the result of the operation should be awaited on.	`False`

Returns:

Type	Description
`Any`	The created entity, or the record of the create task if non_blocking.

`update(body=None, parameters=None, non_blocking=False)` ¶

Update the entity.

Parameters:

Name	Type	Description	Default
`body`	`Optional[EntityDefinition]`	Fields and the values they should be changed to.	`None`
`parameters`	`Optional[Mapping[str, Any]]`	Additional parameters for the call (usually not needed).	`None`
`non_blocking`	`bool`	Whether the result of the operation should be awaited on.	`False`

Returns:

Type	Description
`Any`	The updated entity, or the record of the update task if non_blocking.

`delete(non_blocking=False)` ¶

Delete the entity.

Parameters:

Name	Type	Description	Default
`non_blocking`	`bool`	Whether the result of the operation should be awaited on.	`False`

Returns:

Type	Description
`Any`	The record of the delete task.

`set_label(key, value, non_blocking=False)` ¶

Set a label.

Parameters:

Name	Type	Description	Default
`key`	`str`	Name of the label.	required
`value`	`str`	Value of the label.	required
`non_blocking`	`bool`	Whether the result of the operation should be awaited on. This no longer relevant for pulpcore>=3.34, because the synchronous api is used.	`False`

`unset_label(key, non_blocking=False)` ¶

Unset a label.

Parameters:

Name	Type	Description	Default
`key`	`str`	Name of the label.	required
`non_blocking`	`bool`	Whether the result of the operation should be awaited on. This no longer relevant for pulpcore>=3.34, because the synchronous api is used.	`False`

`show_label(key)` ¶

Show value of a label.

Parameters:

Name	Type	Description	Default
`key`	`str`	Name of the label.	required

Returns:

Type	Description
`Optional[str]`	Value of the label or None.

Raises:

Type	Description
`PulpException`	if the label was not set.

`converge(desired_attributes, defaults=None)` ¶

Converge an entity to have a set of desired attributes.

This will look for the entity, and depending on what it found and what should be, create, delete or update the entity.

Parameters:

Name	Type	Description	Default
`desired_attributes`	`Optional[Dict[str, Any]]`	Dictionary of attributes the entity should have. `None` if the entity is supposed to be absent.	required
`defaults`	`Optional[Dict[str, Any]]`	Optional dict with default and extra values to be used when creating a new entity.	`None`

Returns:

Type	Description
`Tuple[bool, Optional[EntityDefinition], Optional[EntityDefinition]]`	Tuple of (changed, before, after)

`capable(capability)` ¶

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

Parameters:

Name	Type	Description	Default
`capability`	`str`	Name of a capability.	required

Returns:

Type	Description
`bool`	Whether the capability is provided for this context.

`needs_capability(capability)` ¶

Translates a capability in calls to needs_plugin via CAPABILITIES.

Parameters:

Name	Type	Description	Default
`capability`	`str`	Name of a capability.	required

`PulpRemoteContext(pulp_ctx, pulp_href=None, entity=None)` ¶

Bases: PulpEntityContext

Base class for remote contexts.

`PulpPublicationContext(pulp_ctx, pulp_href=None, entity=None)` ¶

Bases: PulpEntityContext

Base class for publication contexts.

`PulpDistributionContext(pulp_ctx, pulp_href=None, entity=None)` ¶

Bases: PulpEntityContext

Base class for distribution contexts.

`PulpRepositoryVersionContext(pulp_ctx, repository_ctx, pulp_href=None)` ¶

Bases: PulpEntityContext

Base class for repository version contexts.

Parameters:

Name	Type	Description	Default
`pulp_ctx`	`PulpContext`	The server context to attach this entity to.	required
`repository_ctx`	`PulpRepositoryContext`	Context of the repository this context should be scoped to.	required
`pulp_href`	`Optional[str]`	Specifying this is equivalent to assinging to `pulp_href` later.	`None`

`repair()` ¶

Trigger a repair task for this repository version.

Returns:

Type	Description
`Any`	The record of the repair task.

`PulpRepositoryContext(pulp_ctx, pulp_href=None, entity=None)` ¶

Bases: PulpEntityContext

Base class for repository contexts.

`get_version_context(number=None)` ¶

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

Parameters:

Name	Type	Description	Default
`number`	`Optional[int]`	Version number or `-1` for the latest version.	`None`

Returns:

Type	Description
`PulpRepositoryVersionContext`	Repository version context attached to the same pulp_ctx and scoped to the repository.

`sync(body=None)` ¶

Trigger a sync task for this repository.

Parameters:

Name	Type	Description	Default
`body`	`Optional[EntityDefinition]`	Any additional options specific to the repository type used to perform this sync.	`None`

Returns:

Type	Description
`Any`	Record of the sync task.

`modify(add_content=None, remove_content=None, base_version=None)` ¶

Add to or remove content from this repository.

Parameters:

Name	Type	Description	Default
`add_content`	`Optional[List[str]]`	List of content hrefs to add.	`None`
`remove_content`	`Optional[List[str]]`	List of content hrefs to remove.	`None`
`base_version`	`Optional[str]`	Href to a repository version relative to whose content the changes are to be interpreted.	`None`

Returns:

Type	Description
`Any`	Record of the modify task.

`PulpGenericRepositoryContext(pulp_ctx, pulp_href=None, entity=None)` ¶

Bases: PulpRepositoryContext

Generic repository context class to separate specific general functionality.

`reclaim(repo_hrefs, repo_versions_keeplist=None)` ¶

Reclaim disk space for a list of repositories.

Parameters:

Name	Type	Description	Default
`repo_hrefs`	`List[Union[str, PulpRepositoryContext]]`	List of repository hrefs to reclaim.	required
`repo_versions_keeplist`	`Optional[List[Union[str, PulpRepositoryVersionContext]]]`	List of repository version hrefs to keep unaffected.	`None`

Returns:

Type	Description
`Any`	Record of the reclaim space task.

`PulpContentContext(pulp_ctx, pulp_href=None, entity=None, repository_ctx=None)` ¶

Bases: PulpEntityContext

Base class for content contexts.

`upload(file, chunk_size, repository, **kwargs)` ¶

Create a content unit by uploading a file.

This function is deprecated. The create call can handle the upload logic transparently.

Parameters:

Name	Type	Description	Default
`file`	`IO[bytes]`	A file like object that supports `os.path.getsize`.	required
`chunk_size`	`int`	Size of the chunks to upload independently.	required
`repository`	`Optional[PulpRepositoryContext]`	Repository context to add the newly created content to.	required
`kwargs`	`Any`	Extra args specific to the content type, passed to the create call.	`{}`

Returns:

Type	Description
`Any`	The result of the create task.

`PulpACSContext(pulp_ctx, pulp_href=None, entity=None)` ¶

Bases: PulpEntityContext

Base class for ACS contexts.

`PulpContentGuardContext(pulp_ctx, pulp_href=None, entity=None)` ¶

Bases: PulpEntityContext

Base class for content guard contexts.

`api_quirk(req)` ¶

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

Parameters:

Name	Type	Description	Default
`req`	`PluginRequirement`	The plugin specifier to determine when the quirk should be applied.	required

Examples:

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

pulp_glue.common.context¶