pulp_glue.core.context

PulpUploadContext

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

Bases: PulpEntityContext

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_file

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

Upload a file and return the uncommitted upload_href.

Source code in pulp-glue/pulp_glue/core/context.py
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
def upload_file(self, file: IO[bytes], chunk_size: int = 1000000) -> 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