pulp_glue.core.context¶

PulpUploadContext ¶

PulpUploadContext(
    pulp_ctx: PulpContext,
    pulp_href: t.Optional[str] = None,
    entity: t.Optional[EntityDefinition] = None,
)

Bases: PulpEntityContext

Source code in pulp-glue/pulp_glue/common/context.py

def __init__(
    self,
    pulp_ctx: PulpContext,
    pulp_href: t.Optional[str] = None,
    entity: t.Optional[EntityDefinition] = None,
) -> None:
    assert pulp_href is None or entity is None

    self.meta: t.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
    self._entity_lookup = entity or {}
    if pulp_href is not None:
        self.pulp_href = pulp_href

upload_file ¶

upload_file(
    file: t.IO[bytes], chunk_size: int = 1000000
) -> t.Any

Upload a file and return the uncommitted upload_href.

Source code in pulp-glue/pulp_glue/core/context.py

def upload_file(self, file: t.IO[bytes], chunk_size: int = 1000000) -> t.Any:
    """Upload a file and return the uncommitted upload_href."""
    start = 0
    size = os.path.getsize(file.name)
    upload_href = self.create(body={"size": size})["pulp_href"]
    try:
        self.pulp_href = upload_href
        while start < size:
            chunk = file.read(chunk_size)
            self.upload_chunk(
                chunk=chunk,
                size=size,
                start=start,
            )
            start += chunk_size
            self.pulp_ctx.echo(".", nl=False, err=True)
    except Exception as e:
        self.delete(upload_href)
        raise e
    self.pulp_ctx.echo(_("Upload complete."), err=True)
    return upload_href