Skip to content

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, timeout, 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.

required
timeout Union[int, timedelta]

Limit of time to wait for unfinished tasks.

required
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 needs to be implemented.

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.

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(limit, offset, parameters)

List entities by the type of this context.

Parameters:

Name Type Description Default
limit int

Maximal number of entities to return.

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)

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.

required

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)

Bases: PulpEntityContext

Base class for content contexts.

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

Create a content unit by uploading a file.

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