spack.oci package


spack.oci.image module

class spack.oci.image.Digest(*, algorithm: str, digest: str)[source]

Bases: object

Represents a digest in the format <algorithm>:<digest>. Currently only supports sha256 digests.

classmethod from_sha256(digest: str) Digest[source]
classmethod from_string(string: str) Digest[source]
class spack.oci.image.ImageReference(*, domain: str, name: str, tag: str = 'latest', digest: Digest | None = None)[source]

Bases: object

A parsed image of the form domain/name:tag[@digest]. The digest is optional, and domain and tag are automatically filled out with defaults when parsed from string.

blob_url(digest: str | Digest) str[source]
endpoint(path: str = '') str[source]
classmethod from_string(string) ImageReference[source]
manifest_url() str[source]
tags_url() str[source]
uploads_url(digest: Digest | None = None) str[source]
with_digest(digest: str | Digest) ImageReference[source]
with_tag(tag: str) ImageReference[source]
spack.oci.image.default_config(architecture: str, os: str)[source]
spack.oci.image.default_index_tag = 'index.spack'

Default OCI index tag

spack.oci.image.default_tag(spec: Spec) str[source]

Return a valid, default image tag for a spec.

spack.oci.image.ensure_sha256_checksum(oci_blob: str)[source]

Validate that the reference is of the format sha256:<checksum> Return the checksum if valid, raise ValueError otherwise.

spack.oci.image.tag_is_spec(tag: str) bool[source]

Check if a tag is likely a Spec

spack.oci.oci module

class spack.oci.oci.Blob(compressed_digest, uncompressed_digest, size)[source]

Bases: NamedTuple

compressed_digest: Digest

Alias for field number 0

size: int

Alias for field number 2

uncompressed_digest: Digest

Alias for field number 1

spack.oci.oci.all_content_type = ['application/vnd.oci.image.manifest.v1+json', 'application/vnd.docker.distribution.manifest.v2+json', 'application/vnd.oci.image.index.v1+json', 'application/vnd.docker.distribution.manifest.list.v2+json']

All OCI manifest / index content types

spack.oci.oci.blob_exists(ref: ImageReference, digest: Digest, _urlopen: Callable[[...], HTTPResponse] | None = None) bool[source]

Checks if a blob exists in an OCI registry

spack.oci.oci.copy_missing_layers(src: ImageReference, dst: ImageReference, architecture: str, _urlopen: Callable[[...], HTTPResponse] | None = None) Tuple[dict, dict][source]

Copy image layers from src to dst for given architecture.

  • src – The source image reference.

  • dst – The destination image reference.

  • architecture – The architecture (when referencing an index)


Tuple of manifest and config of the base image.

spack.oci.oci.copy_missing_layers_with_retry(*args, **kwargs)

Same as copy_missing_layers, but with retry wrapper

spack.oci.oci.create_tarball(spec: Spec, tarfile_path)[source]
spack.oci.oci.get_manifest_and_config(ref: ImageReference, architecture='amd64', recurse=3, _urlopen: Callable[[...], HTTPResponse] | None = None) Tuple[dict, dict][source]

Recursively fetch manifest and config for a given image reference with a given architecture.

  • ref – The image reference.

  • architecture – The architecture (when referencing an index)

  • recurse – How many levels of index to recurse into.


A tuple of (manifest, config)

spack.oci.oci.get_manifest_and_config_with_retry(*args, **kwargs)

Same as get_manifest_and_config, but with retry wrapper

spack.oci.oci.image_from_mirror(mirror: Mirror) ImageReference[source]

Given an OCI based mirror, extract the URL and image name from it

spack.oci.oci.index_content_type = ['application/vnd.oci.image.index.v1+json', 'application/vnd.docker.distribution.manifest.list.v2+json']

OCI index content types (including docker type)

spack.oci.oci.make_stage(url: str, digest: Digest, keep: bool = False, _urlopen: Callable[[...], HTTPResponse] | None = None) Stage[source]
spack.oci.oci.manifest_content_type = ['application/vnd.oci.image.manifest.v1+json', 'application/vnd.docker.distribution.manifest.v2+json']

OCI manifest content types (including docker type)

spack.oci.oci.upload_blob(ref: ImageReference, file: str, digest: Digest, force: bool = False, small_file_size: int = 0, _urlopen: Callable[[...], HTTPResponse] | None = None) bool[source]

Uploads a blob to an OCI registry

We only do monolithic uploads, even though it’s very simple to do chunked. Observed problems with chunked uploads: (1) it’s slow, many sequential requests, (2) some registries set an unknown max chunk size, and the spec doesn’t say how to obtain it

  • ref – The image reference.

  • file – The file to upload.

  • digest – The digest of the file.

  • force – Whether to force upload the blob, even if it already exists.

  • small_file_size – For files at most this size, attempt to do a single POST request instead of POST + PUT. Some registries do no support single requests, and others do not specify what size they support in single POST. For now this feature is disabled by default (0KB)


True if the blob was uploaded, False if it already existed.

spack.oci.oci.upload_blob_with_retry(*args, **kwargs)

Same as upload_blob, but with retry wrapper

spack.oci.oci.upload_manifest(ref: ImageReference, manifest: dict, tag: bool = True, _urlopen: Callable[[...], HTTPResponse] | None = None)[source]

Uploads a manifest/index to a registry

  • ref – The image reference.

  • manifest – The manifest or index.

  • tag – When true, use the tag, otherwise use the digest, this is relevant for multi-arch images, where the tag is an index, referencing the manifests by digest.


The digest and size of the uploaded manifest.

spack.oci.oci.upload_manifest_with_retry(*args, **kwargs)

Same as upload_manifest, but with retry wrapper

spack.oci.oci.with_query_param(url: str, param: str, value: str) str[source]

Add a query parameter to a URL

  • url – The URL to add the parameter to.

  • param – The parameter name.

  • value – The parameter value.


The URL with the parameter added.

spack.oci.opener module

All the logic for OCI fetching and authentication

class spack.oci.opener.Challenge(scheme: str | None = None, params: List[Tuple[str, str]] | None = None)[source]

Bases: object

class spack.oci.opener.OCIAuthHandler(credentials_provider: Callable[[str], UsernamePassword | None])[source]

Bases: BaseHandler

http_error_401(req: Request, fp, code, msg, headers)[source]
https_request(req: Request)[source]
obtain_bearer_token(registry: str, challenge: RealmServiceScope, timeout) str[source]
class spack.oci.opener.RealmServiceScope(realm, service, scope)[source]

Bases: NamedTuple

realm: str

Alias for field number 0

scope: str

Alias for field number 2

service: str

Alias for field number 1

class spack.oci.opener.State(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: Enum

class spack.oci.opener.TokenType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)[source]

Bases: TokenBase

ANY = 7
EOF = 6
class spack.oci.opener.UsernamePassword(username, password)[source]

Bases: NamedTuple

password: str

Alias for field number 1

username: str

Alias for field number 0


Create an opener that can handle OCI authentication.

spack.oci.opener.credentials_from_mirrors(domain: str, *, mirrors: Iterable[Mirror] | None = None) UsernamePassword | None[source]

Filter out OCI registry credentials from a list of mirrors.

spack.oci.opener.default_retry(f, retries: int = 3, sleep=None)[source]
spack.oci.opener.ensure_status(request: Request, response: HTTPResponse, status: int)[source]

Raise an error if the response status is not the expected one.

spack.oci.opener.get_bearer_challenge(challenges: List[Challenge]) RealmServiceScope | None[source]
spack.oci.opener.parse_www_authenticate(input: str)[source]

Very basic parsing of www-authenticate parsing (RFC7235 section 4.1) Notice: this omits token68 support.

spack.oci.opener.tokenize(input: str)[source]
spack.oci.opener.urlopen: Callable[[...], HTTPResponse] = <function _urlopen.<locals>.dispatch_open>

Opener that automatically uses OCI authentication based on mirror config