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

`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, verify=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, default: False ) –

Whether to wait for tasks. If True, all tasks triggered will immediately raise PulpNoWait.
timeout (Union[int, timedelta], default: 300 ) –

Limit of time to wait for unfinished tasks.
domain (str, default: 'default' ) –

Name of the domain to interact with.
fake_mode (bool, default: False ) –

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.
verify (Optional[Union[bool, str]], default: None ) –

A boolean or a path to the CA bundle.

`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:

profile (Optional[str], default: None ) –

Select a different profile from the config.
config_locations (Optional[List[str]], default: None ) –

If provided these config files will be merged (last on wins) instead of the default locations.

Returns: A configured PulpContext object.

`from_config(config)` `classmethod` ¶

Create a PulpContext object from a config dictionary.

Parameters:

config (Dict[str, Any]) –

dictionary of configuration values.

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:

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.
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:

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.

`wait_for_task_group(task_group)` ¶

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.

`PulpViewSetContext(pulp_ctx)` ¶

Base class to interact with a generic viewset.

Parameters:

pulp_ctx (PulpContext) –

The server context to attach this viewset context to.

`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:

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.

`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:

pulp_ctx (PulpContext) –

The server context to attach this entity context 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.

`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:

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.

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

List entities from this context in a batched iterator.

Parameters:

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

Search or sorting criteria.
offset (int, default: 0 ) –

Number of entities to skip.
batch_size (int, default: BATCH_SIZE ) –

Size of the batches to fetch. Maximally BATCH_SIZE will be used.
stats (Optional[Dict[str, Any]], default: None ) –

If provided, a dictionary that will be filled with metadata: count: Number of entities reported by the server to match the criteria.

Returns:

Iterator[Any] –

Iterator of entities matching the search conditions.

`list(limit, offset, parameters)` ¶

List entities by the type of this context.

Parameters:

limit (int) –

Maximal number of entities to return Use 0 to loop until all entries are retrieved.
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.

`find(**kwargs)` ¶

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:

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:

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.

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

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.

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

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.

Returns:

Any –

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

`delete(non_blocking=False)` ¶

Delete the entity.

Parameters:

non_blocking (bool, default: False ) –

Whether the result of the operation should be awaited on.

Returns:

Any –

The record of the delete task.

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

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.

`unset_label(key, non_blocking=False)` ¶

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.

`show_label(key)` ¶

Show value of a label.

Parameters:

key (str) –

Name of the label.

Returns:

Optional[str] –

Value of the label or None.

Raises:

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:

desired_attributes (Optional[Dict[str, Any]]) –

Dictionary of attributes the entity should have. None if the entity is supposed to be absent.
defaults (Optional[Dict[str, Any]], default: None ) –

Optional dict with default and extra values to be used when creating a new entity.

Returns:

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:

capability (str) –

Name of a capability.

Returns:

bool –

Whether the capability is provided for this context.

`needs_capability(capability)` ¶

Translates a capability in calls to needs_plugin via CAPABILITIES.

Parameters:

capability (str) –

Name of a capability.

`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:

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.

`repair()` ¶

Trigger a repair task for this repository version.

Returns:

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:

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

Version number or -1 for the latest version.

Returns:

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:

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

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

Returns:

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:

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.

Returns:

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:

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.

`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:

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.

`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:

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

pulp_glue.common.context¶