spack.util package¶

Subpackages¶

spack.util.unparse package

Submodules¶

spack.util.archive module¶

class spack.util.archive.ChecksumWriter(fileobj, algorithm=hashlib.sha256)[source]¶

Bases: BufferedIOBase

Checksum writer computes a checksum while writing to a file.

close()[source]¶

Flush and close the IO object.

This method has no effect if the file is already closed.

property closed¶

fileno()[source]¶

Return underlying file descriptor if one exists.

Raise OSError if the IO object does not use a file descriptor.

flush()[source]¶

Flush write buffers, if applicable.

This is not implemented for read-only and non-blocking streams.

hexdigest()[source]¶

myfileobj¶

peek(n)[source]¶

read(size=-1)[source]¶

Read and return up to n bytes.

If the size argument is omitted, None, or negative, read and return all data until EOF.

If the size argument is positive, and the underlying raw stream is not ‘interactive’, multiple raw reads may be issued to satisfy the byte count (unless EOF is reached first). However, for interactive raw streams (as well as sockets and pipes), at most one raw read will be issued, and a short result does not imply that EOF is imminent.

Return an empty bytes object on EOF.

Return None if the underlying raw stream was open in non-blocking mode and no data is available at the moment.

read1(size=-1)[source]¶

Read and return up to size bytes, with at most one read() call to the underlying raw stream.

Return an empty bytes object on EOF. A short result does not imply that EOF is imminent.

readable()[source]¶

Return whether object was opened for reading.

If False, read() will raise OSError.

readline(size=-1)[source]¶

Read and return a line from the stream.

If size is specified, at most size bytes will be read.

The line terminator is always b’n’ for binary files; for text files, the newlines argument to open can be used to select the line terminator(s) recognized.

rewind()[source]¶

seek(offset, whence=io.SEEK_SET)[source]¶

Change the stream position to the given byte offset.

offset
The stream position, relative to ‘whence’.

whence
The relative position to seek from.

The offset is interpreted relative to the position indicated by whence. Values for whence are:

os.SEEK_SET or 0 – start of stream (the default); offset should be zero or positive
os.SEEK_CUR or 1 – current stream position; offset may be negative
os.SEEK_END or 2 – end of stream; offset is usually negative

Return the new absolute position.

seekable()[source]¶

Return whether object supports random access.

If False, seek(), tell() and truncate() will raise OSError. This method may need to do a test seek().

tell()[source]¶: Return current stream position.

writable()[source]¶

Return whether object was opened for writing.

If False, write() will raise OSError.

write(data)[source]¶

Write buffer b to the IO stream.

Return the number of bytes written, which is always the length of b in bytes.

Raise BlockingIOError if the buffer is full and the underlying raw stream cannot accept more data at the moment.

spack.util.archive.default_add_file(tar: TarFile, file_info: TarInfo, path: str) → None[source]¶

spack.util.archive.default_add_link(tar: TarFile, file_info: TarInfo, path: str) → None[source]¶

spack.util.archive.default_path_to_name(path: str) → str[source]¶: Converts a path to a tarfile name, which uses posix path separators.

spack.util.archive.gzip_compressed_tarfile(path: str) → Generator[Tuple[TarFile, ChecksumWriter, ChecksumWriter], None, None][source]¶

Create a reproducible, gzip compressed tarfile, and keep track of shasums of both the compressed and uncompressed tarfile. Reproducibility is achieved by normalizing the gzip header (no file name and zero mtime).

Yields:

A tuple of three elements

tarfile.TarFile: tarfile object
ChecksumWriter: checksum of the gzip compressed tarfile
ChecksumWriter: checksum of the uncompressed tarfile

spack.util.archive.reproducible_tarfile_from_prefix(tar: TarFile, prefix: str, *, include_parent_directories: bool = False, skip: Callable[[DirEntry], bool] = lambda entry: ..., path_to_name: Callable[[str], str] = default_path_to_name, add_file: Callable[[TarFile, TarInfo, str], None] = default_add_file, add_symlink: Callable[[TarFile, TarInfo, str], None] = default_add_link, add_hardlink: Callable[[TarFile, TarInfo, str], None] = default_add_link) → None[source]¶

Create a tarball from a given directory. Only adds regular files, symlinks and dirs. Skips devices, fifos. Preserves hardlinks. Normalizes permissions like git. Tar entries are added in depth-first pre-order, with dir entries partitioned by file | dir, and sorted lexicographically, for reproducibility. Partitioning ensures only one dir is in memory at a time, and sorting improves compression.

Parameters:

tar – tarfile object opened in write mode
prefix – path to directory to tar (either absolute or relative)
include_parent_directories – whether to include every directory leading up to prefix in the tarball
skip – function that receives a DirEntry and returns True if the entry should be skipped, whether it is a file or directory. Default implementation does not skip anything.
path_to_name – function that converts a path string to a tarfile entry name, which should be in posix format. Not only is it necessary to transform paths in certain cases, such as windows path to posix format, but it can also be used to prepend a directory to each entry even if it does not exist on the filesystem. The default implementation drops the leading slash on posix and the drive letter on windows for absolute paths, and formats as a posix.

spack.util.archive.retrieve_commit_from_archive(archive_path, ref)[source]¶

Extract git data from an archive with out expanding it

Open the archive and searches for .git/HEAD. Return if HEAD is a commit (detached head or tag) Otherwise attempt to read the ref that .git/HEAD is pointing to and return the commit associated with it.

spack.util.argparsewriter module¶

class spack.util.argparsewriter.ArgparseRstWriter(prog: str, out: IO = sys.stdout, aliases: bool = False, rst_levels: Sequence[str] = _rst_levels)[source]¶

Bases: ArgparseWriter

Write argparse output as rst sections.

begin_command(prog: str) → str[source]¶

Text to print before a command.

Parameters:: prog – Program name.
Returns:: Text before a command.

begin_optionals() → str[source]¶

Text to print before optional arguments.

Returns:: Optional arguments header.

begin_positionals() → str[source]¶

Text to print before positional arguments.

Returns:: Positional arguments header.

begin_subcommands(subcommands: List[Tuple[ArgumentParser, str, str]]) → str[source]¶

Table with links to other subcommands.

Parameters:: subcommands – List of subcommands.
Returns:: Subcommand linking text.

description(description: str) → str[source]¶

Description of a command.

Parameters:: description – Command description.
Returns:: Description of a command.

end_optionals() → str[source]¶

Text to print after optional arguments.

Returns:: Optional arguments footer.

end_positionals() → str[source]¶

Text to print after positional arguments.

Returns:: Positional arguments footer.

format(cmd: Command) → str[source]¶

Return the string representation of a single node in the parser tree.

Parameters:: cmd – Parsed information about a command or subcommand.
Returns:: String representation of a node.

optional(opts: str, help: str) → str[source]¶

Description of an optional argument.

Parameters:

opts – Optional argument.
help – Help text.

Returns:

Optional argument description.

positional(name: str, help: str) → str[source]¶

Description of a positional argument.

Parameters:

name – Argument name.
help – Help text.

Returns:

Positional argument description.

usage(usage: str) → str[source]¶

Example usage of a command.

Parameters:: usage – Command usage.
Returns:: Usage of a command.

class spack.util.argparsewriter.ArgparseWriter(prog: str, out: IO = sys.stdout, aliases: bool = False)[source]¶

Bases: HelpFormatter, ABC

Analyze an argparse ArgumentParser for easy generation of help.

abstractmethod format(cmd: Command) → str[source]¶

Return the string representation of a single node in the parser tree.

Override this in subclasses to define how each subcommand should be displayed.

Parameters:: cmd – Parsed information about a command or subcommand.
Returns:: String representation of this subcommand.

parse(parser: ArgumentParser, prog: str) → Command[source]¶

Parse the parser object and return the relevant components.

Parameters:

parser – Command parser.
prog – Program name.

Returns:

Information about the command from the parser.

write(parser: ArgumentParser) → None[source]¶

Write out details about an ArgumentParser.

Parameters:: parser – Command parser.

class spack.util.argparsewriter.Command(prog: str, description: str | None, usage: str, positionals: List[Tuple[str, Iterable[Any] | None, int | str | None, str]], optionals: List[Tuple[Sequence[str], List[str], str, int | str | None, str]], subcommands: List[Tuple[ArgumentParser, str, str]])[source]¶

Bases: object

Parsed representation of a command from argparse.

This is a single command from an argparse parser. ArgparseWriter creates these and returns them from parse(), and it passes one of these to each call to format() so that we can take an action for a single command.

spack.util.compression module¶

class spack.util.compression.BZipFileType[source]¶

Bases: CompressedFileTypeInterface

extension: str¶

name: str¶

peek(stream: BinaryIO, num_bytes: int) → BytesIO | None[source]¶: This method returns the first num_bytes of a decompressed stream. Returns None if no builtin support for decompression.

class spack.util.compression.CompressedFileTypeInterface[source]¶

Bases: FileTypeInterface

Interface class for FileTypes that include compression information

peek(stream: BinaryIO, num_bytes: int) → BytesIO | None[source]¶: This method returns the first num_bytes of a decompressed stream. Returns None if no builtin support for decompression.

class spack.util.compression.FileTypeInterface[source]¶

Bases: object

Base interface class for describing and querying file type information. FileType describes information about a single file type such as typical extension and byte header properties, and provides an interface to check a given file against said type based on magic number.

This class should be subclassed each time a new type is to be described.

Subclasses should each describe a different type of file. In order to do so, they must define the extension string, magic number, and header offset (if non zero). If a class has multiple magic numbers, it will need to override the method describing that file type’s magic numbers and the method that checks a types magic numbers against a given file’s.

OFFSET¶

extension: str¶

classmethod header_size() → int[source]¶: Return size of largest magic number associated with file type

classmethod magic_numbers() → List[bytes][source]¶: Return a list of all potential magic numbers for a filetype

matches_magic(stream: BinaryIO) → bool[source]¶

Returns true if the stream matches the current file type by any of its magic numbers. Resets stream to original position.

Parameters:: stream – file byte stream

name: str¶

class spack.util.compression.GZipFileType[source]¶

Bases: CompressedFileTypeInterface

extension: str¶

name: str¶

peek(stream: BinaryIO, num_bytes: int) → BytesIO | None[source]¶: This method returns the first num_bytes of a decompressed stream. Returns None if no builtin support for decompression.

class spack.util.compression.LzmaFileType[source]¶

Bases: CompressedFileTypeInterface

extension: str¶

name: str¶

peek(stream: BinaryIO, num_bytes: int) → BytesIO | None[source]¶: This method returns the first num_bytes of a decompressed stream. Returns None if no builtin support for decompression.

spack.util.compression.MAX_BYTES_ARCHIVE_HEADER¶: Maximum number of bytes to read from a file to determine any archive type. Tar is the largest.

spack.util.compression.SUPPORTED_FILETYPES: List[FileTypeInterface]¶: Collection of supported archive and compression file type identifier classes.

class spack.util.compression.TarFileType[source]¶

Bases: FileTypeInterface

OFFSET¶

extension: str¶

name: str¶

class spack.util.compression.ZCompressedFileType[source]¶

Bases: CompressedFileTypeInterface

extension: str¶

name: str¶

class spack.util.compression.ZipFleType[source]¶

Bases: FileTypeInterface

extension: str¶

name: str¶

spack.util.compression.decompressor_for(path: str, extension: str | None = None)[source]¶: Returns appropriate decompression/extraction algorithm function pointer for provided extension. If extension is none, it is computed from the path and the decompression function is derived from that information.

spack.util.compression.decompressor_for_nix(extension: str) → Callable[[str], Any][source]¶

Returns a function pointer to appropriate decompression algorithm based on extension type and unix specific considerations i.e. a reasonable expectation system utils like gzip, bzip2, and xz are available

Parameters:: extension – path of the archive file requiring decompression

spack.util.compression.decompressor_for_win(extension: str) → Callable[[str], Any][source]¶

Returns a function pointer to appropriate decompression algorithm based on extension type and Windows specific considerations

Windows natively vendors only tar, no other archive/compression utilities So we must rely exclusively on Python module support for all compression operations, tar for tarballs and zip files, and 7zip for Z compressed archives and files as Python does not provide support for the UNIX compress algorithm

spack.util.compression.extension_from_magic_numbers(path: str, decompress: bool = False) → str | None[source]¶

Return typical extension without leading . of a compressed file or archive at the given path, based on its magic numbers, similar to the file utility. Notice that the extension returned from this function may not coincide with the file’s given extension.

Parameters:

path – file to determine extension of
decompress – If True, method will peek into decompressed file to check for archive file types. If False, the method will return only the top-level extension (for example gz and not tar.gz).

Returns:

Spack recognized archive file extension as determined by file’s magic number and file name. If file is not on system or is of a type not recognized by Spack as an archive or compression type, None is returned. If the file is classified as a compressed tarball, the extension is abbreviated (for instance tgz not tar.gz) if that matches the file’s given extension.

spack.util.compression.extension_from_magic_numbers_by_stream(stream: BinaryIO, decompress: bool = False) → str | None[source]¶

Returns the typical extension for the opened file, without leading ., based on its magic numbers.

If the stream does not represent file type recognized by Spack (see SUPPORTED_FILETYPES), the method will return None

Parameters:

stream – stream representing a file on system
decompress – if True, compressed files are checked for archive types beneath compression. For example tar.gz if True versus only gz if False.

spack.util.cpus module¶

spack.util.cpus.cpus_available()[source]¶: Returns the number of CPUs available for the current process, or the number of physical CPUs when that information cannot be retrieved. The number of available CPUs might differ from the number of physical CPUs when using spack through Slurm or container runtimes.

spack.util.crypto module¶

class spack.util.crypto.Checker(hexdigest: str, **kwargs)[source]¶

Bases: object

A checker checks files against one particular hex digest. It will automatically determine what hashing algorithm to used based on the length of the digest it’s initialized with. e.g., if the digest is 32 hex characters long this will use md5.

Example: know your tarball should hash to abc123. You want to check files against this. You would use this class like so:

hexdigest = 'abc123'
checker = Checker(hexdigest)
success = checker.check('downloaded.tar.gz')

After the call to check, the actual checksum is available in checker.sum, in case it’s needed for error output.

You can trade read performance and memory usage by adjusting the block_size optional arg. By default it’s a 1MB (2**20 bytes) buffer.

check(filename: str) → bool[source]¶: Read the file with the specified name and check its checksum against self.hexdigest. Return True if they match, False otherwise. Actual checksum is stored in self.sum.

property hash_name: str¶: Get the name of the hash function this Checker is using.

sum: str | None¶

class spack.util.crypto.DeprecatedHash(hash_alg, alert_fn, disable_security_check)[source]¶: Bases: object

spack.util.crypto.bit_length(num)[source]¶: Number of bits required to represent an integer in binary.

spack.util.crypto.checksum(hashlib_algo: Callable[[], hashlib._Hash], filename: str, *, block_size: int = 2**20) → str[source]¶: Returns a hex digest of the filename generated using an algorithm from hashlib.

spack.util.crypto.checksum_stream(hashlib_algo: Callable[[], hashlib._Hash], fp: BinaryIO, *, block_size: int = 2**20) → str[source]¶: Returns a hex digest of the stream generated using given algorithm from hashlib.

spack.util.crypto.hash_algo_for_digest(hexdigest: str) → str[source]¶: Gets name of the hash algorithm for a hex digest.

spack.util.crypto.hash_fun_for_algo(algo: str) → Callable[[], hashlib._Hash][source]¶: Get a function that can perform the specified hash algorithm.

spack.util.crypto.hash_fun_for_digest(hexdigest: str) → Callable[[], hashlib._Hash][source]¶: Gets a hash function corresponding to a hex digest.

spack.util.crypto.hashes¶: Set of hash algorithms that Spack can use, mapped to digest size in bytes

spack.util.crypto.prefix_bits(byte_array, bits)[source]¶: Return the first <bits> bits of a byte array as an integer.

spack.util.ctest_log_parser module¶

Functions to parse build logs and extract error messages.

This is a python port of the regular expressions CTest uses to parse log files here:

https://github.com/Kitware/CMake/blob/master/Source/CTest/cmCTestBuildHandler.cxx

This file takes the regexes verbatim from there and adds some parsing algorithms that duplicate the way CTest scrapes log files. To keep this up to date with CTest, just make sure the *_matches and *_exceptions lists are kept up to date with CTest’s build handler.

class spack.util.ctest_log_parser.BuildError(text: str, line_no: int, source_file: str | None = None, source_line_no: str | None = None, pre_context: List[str] | None = None, post_context: List[str] | None = None)[source]¶

Bases: LogEvent

LogEvent subclass for build errors.

color¶: color name when rendering in the terminal

class spack.util.ctest_log_parser.BuildWarning(text: str, line_no: int, source_file: str | None = None, source_line_no: str | None = None, pre_context: List[str] | None = None, post_context: List[str] | None = None)[source]¶

Bases: LogEvent

LogEvent subclass for build warnings.

color¶: color name when rendering in the terminal

class spack.util.ctest_log_parser.CTestLogParser(profile: bool = False)[source]¶

Bases: object

Log file parser that extracts errors and warnings.

parse(stream: str | TextIO | List[str], context: int = 6, tail: int = 0) → Tuple[List[BuildError], List[BuildWarning], LogEvent | None][source]¶

Parse a log file by searching each line for errors and warnings.

Parameters:

stream – filename or stream to read from
context – lines of context to extract around each log event
tail – if > 0, also return a LogEvent with the last tail lines

Returns:

two lists containing BuildError and BuildWarning objects, plus an optional LogEvent for the tail (None when tail=0).

print_timings() → None[source]¶: Print out profile of time spent in different regular expressions.

class spack.util.ctest_log_parser.LogEvent(text: str, line_no: int, source_file: str | None = None, source_line_no: str | None = None, pre_context: List[str] | None = None, post_context: List[str] | None = None)[source]¶

Bases: object

Class representing interesting events (e.g., errors) in a build log.

color¶: color name when rendering in the terminal

property end: int¶: Last line in the log with text for event or its context.

property start: int¶: First line in the log with text for the event or its context.

spack.util.editor module¶

Module for finding the user’s preferred text editor.

Defines one function, editor(), which invokes the editor defined by the user’s VISUAL environment variable if set. We fall back to the editor defined by the EDITOR environment variable if VISUAL is not set or the specified editor fails (e.g. no DISPLAY for a graphical editor). If neither variable is set, we fall back to one of several common editors, raising an OSError if we are unable to find one.

spack.util.editor.editor(*args: str, exec_fn: Callable[[str, List[str]], int] = os.execv) → bool[source]¶

Invoke the user’s editor.

This will try to execute the following, in order:

$VISUAL <args>: the “visual” editor (per POSIX)
$EDITOR <args>: the regular editor (per POSIX)
some default editor (see _default_editors) with <args>

If an environment variable isn’t defined, it is skipped. If it points to something that can’t be executed, we’ll print a warning. And if we can’t find anything that can be executed after searching the full list above, we’ll raise an error.

Parameters:

args – args to pass to editor
exec_fn – invoke this function to run; use spack.util.editor.executable if you want something that returns, instead of the default os.execv().

spack.util.editor.executable(exe: str, args: List[str]) → int[source]¶

Wrapper that makes spack.util.executable.Executable look like os.execv().

Use this with editor() if you want it to return instead of running execv.

spack.util.elf module¶

class spack.util.elf.CStringType[source]¶

Bases: object

PT_INTERP¶

RPATH¶

class spack.util.elf.ELF_CONSTANTS[source]¶

Bases: object

CLASS32¶

CLASS64¶

DATA2LSB¶

DATA2MSB¶

DT_NEEDED¶

DT_NULL¶

DT_RPATH¶

DT_RUNPATH¶

DT_SONAME¶

DT_STRTAB¶

ET_DYN¶

ET_EXEC¶

MAGIC¶

PT_DYNAMIC¶

PT_INTERP¶

PT_LOAD¶

SHT_STRTAB¶

exception spack.util.elf.ElfCStringUpdatesFailed(rpath: UpdateCStringAction | None, pt_interp: UpdateCStringAction | None)[source]¶: Bases: Exception

class spack.util.elf.ElfFile[source]¶

Bases: object

Parsed ELF file.

byte_order: str¶

dt_needed_strs: List[bytes]¶

dt_needed_strtab_offsets: List[int]¶

dt_rpath_offset: int¶

dt_rpath_str: bytes¶

dt_soname_str: bytes¶

dt_soname_strtab_offset: int¶

elf_hdr: ElfHeader¶

has_needed: bool¶

has_pt_dynamic: bool¶

has_pt_interp: bool¶

has_rpath: bool¶

has_soname: bool¶

is_64_bit: bool¶

is_little_endian: bool¶

is_runpath: bool¶

pt_dynamic_p_filesz: int¶

pt_dynamic_p_offset: int¶

pt_dynamic_strtab_offset: int¶

pt_interp_p_filesz: int¶

pt_interp_p_offset: int¶

pt_interp_str: bytes¶

pt_load: List[Tuple[int, int]]¶

rpath_strtab_offset: int¶

class spack.util.elf.ElfHeader(e_type, e_machine, e_version, e_entry, e_phoff, e_shoff, e_flags, e_ehsize, e_phentsize, e_phnum, e_shentsize, e_shnum, e_shstrndx)[source]¶

Bases: NamedTuple

e_ehsize: int¶: Alias for field number 7

e_entry: int¶: Alias for field number 3

e_flags: int¶: Alias for field number 6

e_machine: int¶: Alias for field number 1

e_phentsize: int¶: Alias for field number 8

e_phnum: int¶: Alias for field number 9

e_phoff: int¶: Alias for field number 4

e_shentsize: int¶: Alias for field number 10

e_shnum: int¶: Alias for field number 11

e_shoff: int¶: Alias for field number 5

e_shstrndx: int¶: Alias for field number 12

e_type: int¶: Alias for field number 0

e_version: int¶: Alias for field number 2

exception spack.util.elf.ElfParsingError[source]¶: Bases: Exception

class spack.util.elf.ProgramHeader32(p_type, p_offset, p_vaddr, p_paddr, p_filesz, p_memsz, p_flags, p_align)[source]¶

Bases: NamedTuple

p_align: int¶: Alias for field number 7

p_filesz: int¶: Alias for field number 4

p_flags: int¶: Alias for field number 6

p_memsz: int¶: Alias for field number 5

p_offset: int¶: Alias for field number 1

p_paddr: int¶: Alias for field number 3

p_type: int¶: Alias for field number 0

p_vaddr: int¶: Alias for field number 2

class spack.util.elf.ProgramHeader64(p_type, p_flags, p_offset, p_vaddr, p_paddr, p_filesz, p_memsz, p_align)[source]¶

Bases: NamedTuple

p_align: int¶: Alias for field number 7

p_filesz: int¶: Alias for field number 5

p_flags: int¶: Alias for field number 1

p_memsz: int¶: Alias for field number 6

p_offset: int¶: Alias for field number 2

p_paddr: int¶: Alias for field number 4

p_type: int¶: Alias for field number 0

p_vaddr: int¶: Alias for field number 3

class spack.util.elf.SectionHeader(sh_name, sh_type, sh_flags, sh_addr, sh_offset, sh_size, sh_link, sh_info, sh_addralign, sh_entsize)[source]¶

Bases: NamedTuple

sh_addr: int¶: Alias for field number 3

sh_addralign: int¶: Alias for field number 8

sh_entsize: int¶: Alias for field number 9

sh_flags: int¶: Alias for field number 2

sh_info: int¶: Alias for field number 7

sh_link: int¶: Alias for field number 6

sh_name: int¶: Alias for field number 0

sh_offset: int¶: Alias for field number 4

sh_size: int¶: Alias for field number 5

sh_type: int¶: Alias for field number 1

class spack.util.elf.UpdateCStringAction(old_value: bytes, new_value: bytes, offset: int)[source]¶

Bases: object

apply(f: BinaryIO) → None[source]¶

property inplace: bool¶

spack.util.elf.delete_needed_from_elf(f: BinaryIO, elf: ElfFile, needed: bytes) → None[source]¶: Delete a needed library from the dynamic section of an ELF file

spack.util.elf.delete_rpath(path: str) → None[source]¶: Modifies a binary to remove the rpath. It zeros out the rpath string and also drops the DT_RPATH / DT_RUNPATH entry from the dynamic section, so it doesn’t show up in readelf -d file, nor in strings file.

spack.util.elf.find_strtab_size_at_offset(f: BinaryIO, elf: ElfFile, offset: int) → int[source]¶

Retrieve the size of a string table section at a particular known offset

Parameters:

f – file handle
elf – ELF file parser data
offset – offset of the section in the file (i.e. sh_offset)

Returns:

the size of the string table in bytes

Return type:

int

spack.util.elf.get_elf_compat(path: str) → Tuple[bool, bool, int][source]¶: Get a triplet (is_64_bit, is_little_endian, e_machine) from an ELF file, which can be used to see if two ELF files are compatible.

spack.util.elf.get_interpreter(path: str) → str | None[source]¶: Returns the interpreter of the given file as UTF-8 string, or None if not set.

spack.util.elf.get_rpaths(path: str) → List[str] | None[source]¶: Returns list of rpaths of the given file as UTF-8 strings, or None if not set.

spack.util.elf.parse_c_string(byte_string: bytes, start: int = 0) → bytes[source]¶

Retrieve a C-string at a given offset in a byte string

Parameters:

byte_string – String
start – Offset into the string

Returns:

A copy of the C-string excluding the terminating null byte

Return type:

bytes

spack.util.elf.parse_elf(f: BinaryIO, interpreter: bool = False, dynamic_section: bool = False, only_header: bool = False) → ElfFile[source]¶: Given a file handle f for an ELF file opened in binary mode, return an ElfFile object with the parsed contents.

spack.util.elf.parse_header(f: BinaryIO, elf: ElfFile) → None[source]¶

spack.util.elf.parse_program_headers(f: BinaryIO, elf: ElfFile) → None[source]¶

Parse program headers

Parameters:

f – file handle
elf – ELF file parser data

spack.util.elf.parse_pt_dynamic(f: BinaryIO, elf: ElfFile) → None[source]¶

Parse the dynamic section of an ELF file

Parameters:

f – file handle
elf – ELF file parse data

spack.util.elf.parse_pt_interp(f: BinaryIO, elf: ElfFile) → None[source]¶

Parse the interpreter (i.e. absolute path to the dynamic linker)

Parameters:

f – file handle
elf – ELF file parser data

spack.util.elf.pt_interp(path: str) → str | None[source]¶: Retrieve the interpreter of an executable at path.

spack.util.elf.read_exactly(f: BinaryIO, num_bytes: int, msg: str) → bytes[source]¶

Read exactly num_bytes at the current offset, otherwise raise a parsing error with the given error message.

Parameters:

f – file handle
num_bytes – Number of bytes to read
msg – Error to show when bytes cannot be read

Returns:

the num_bytes bytes that were read.

Return type:

bytes

spack.util.elf.retrieve_strtab(f: BinaryIO, elf: ElfFile, offset: int) → bytes[source]¶

Read a full string table at the given offset, which requires looking it up in the section headers.

Parameters:

elf – ELF file parser data
vaddr – virtual address

Returns: file offset

spack.util.elf.substitute_rpath_and_pt_interp_in_place_or_raise(path: str, substitutions: Dict[bytes, bytes]) → bool[source]¶: Returns true if the rpath and interpreter were modified, false if there was nothing to do. Raises ElfCStringUpdatesFailed if the ELF file cannot be updated in-place. This exception contains a list of actions to perform with other tools. The file is left untouched in this case.

spack.util.elf.vaddr_to_offset(elf: ElfFile, vaddr: int) → int[source]¶

Given a virtual address, find the corresponding offset in the ELF file itself.

Parameters:

elf – ELF file parser data
vaddr – virtual address

spack.util.environment module¶

Set, unset or modify environment variables.

class spack.util.environment.AppendFlagsEnv(name: str, value: str, *, separator: str = os.pathsep, trace: Trace | None = None)[source]¶

Bases: NameValueModifier

execute(env: MutableMapping[str, str])[source]¶: Apply the modification to the mapping passed as input

name¶

separator¶

trace¶

value¶

class spack.util.environment.AppendPath(name: str, value: str | PurePath, *, separator: str = os.pathsep, trace: Trace | None = None)[source]¶

Bases: NamePathModifier

execute(env: MutableMapping[str, str])[source]¶: Apply the modification to the mapping passed as input

class spack.util.environment.DeprioritizeSystemPaths(name: str, *, separator: str = os.pathsep, trace: Trace | None = None)[source]¶

Bases: NameModifier

execute(env: MutableMapping[str, str])[source]¶: Apply the modification to the mapping passed as input

name¶

separator¶

trace¶

class spack.util.environment.EnvironmentModifications(other: EnvironmentModifications | None = None, traced: None | bool = None)[source]¶

Bases: object

Tracks and applies a sequence of environment variable modifications.

This class provides a high-level interface for building up a list of environment changes, such as setting, unsetting, appending, prepending, or removing values from environment variables. Modifications are stored and can be applied to a given environment dictionary, or rendered as shell code.

Package authors typically receive an instance of this class and call set(), unset(), prepend_path(), remove_path(), etc., to queue up modifications. Spack runs apply_modifications() to apply these modifications to the environment when needed.

Modifications can be grouped by variable name, reversed (where possible), validated for suspicious patterns, and extended from other instances. The class also supports tracing the origin of modifications for debugging.

Example

env = EnvironmentModifications()
env.set("FOO", "bar")
env.prepend_path("PATH", "/custom/bin")
env.apply_modifications()  # applies changes to os.environ

append_flags(name: str, value: str, sep: str = ' ') → None[source]¶

Stores a request to append flags to an environment variable.

Parameters:

name – name of the environment variable
value – flags to be appended
sep – separator for the flags (default: " ")

append_path(name: str, path: str | PurePath, separator: str = os.pathsep) → None[source]¶

Stores a request to append a path to list of paths.

Parameters:

name – name of the environment variable
path – path to be appended
separator – separator for the paths (default: os.pathsep)

apply_modifications(env: MutableMapping[str, str] | None = None)[source]¶

Applies the modifications to the environment.

Parameters:: env – environment to be modified. If None, os.environ will be used.

clear()[source]¶: Clears the current list of modifications.

deprioritize_system_paths(name: str, separator: str = os.pathsep) → None[source]¶

Stores a request to deprioritize system paths in a path list, otherwise preserving the order.

Parameters:

name – name of the environment variable
separator – separator for the paths (default: os.pathsep)

drop(*name) → bool[source]¶: Drop all modifications to the variable with the given name.

env_modifications: List[NameModifier | NameValueModifier]¶

extend(other: EnvironmentModifications)[source]¶: Extends the current instance with modifications from another instance.

static from_environment_diff(before: MutableMapping[str, str], after: MutableMapping[str, str], clean: bool = False) → EnvironmentModifications[source]¶

Constructs the environment modifications from the diff of two environments.

Parameters:

before – environment before the modifications are applied
after – environment after the modifications are applied
clean – in addition to removing empty entries, also remove duplicate entries

static from_sourcing_file(filename: str, *arguments: str, **kwargs: Any) → EnvironmentModifications[source]¶

Returns the environment modifications that have the same effect as sourcing the input file in a shell.

Parameters:

filename – the file to be sourced
*arguments – arguments to pass on the command line

Keyword Arguments:

shell (str) – the shell to use (default: bash)
shell_options (str) – options passed to the shell (default: -c)
source_command (str) – the command to run (default: source)
suppress_output (str) – redirect used to suppress output of command (default: &> /dev/null)
concatenate_on_success (str) – operator used to execute a command only when the previous command succeeds (default: &&)
exclude ([str or re.Pattern[str]]) – ignore any modifications of these variables (default: [])
include ([str or re.Pattern[str]]) – always respect modifications of these variables (default: []). Supersedes any excluded variables.
clean (bool) – in addition to removing empty entries, also remove duplicate entries (default: False).

group_by_name() → Dict[str, List[NameModifier | NameValueModifier]][source]¶: Returns a dict of the current modifications keyed by variable name.

is_unset(variable_name: str) → bool[source]¶: Returns True if the last modification to a variable is to unset it, False otherwise.

prepend_path(name: str, path: str | PurePath, separator: str = os.pathsep) → None[source]¶

Stores a request to prepend a path to list of paths.

Parameters:

name – name of the environment variable
path – path to be prepended
separator – separator for the paths (default: os.pathsep)

prune_duplicate_paths(name: str, separator: str = os.pathsep) → None[source]¶

Stores a request to remove duplicates from a path list, otherwise preserving the order.

Parameters:

name – name of the environment variable
separator – separator for the paths (default: os.pathsep)

remove_first_path(name: str, path: str | PurePath, separator: str = os.pathsep) → None[source]¶

Stores a request to remove first instance of path from a list of paths.

Parameters:

name – name of the environment variable
path – path to be removed
separator – separator for the paths (default: os.pathsep)

remove_flags(name: str, value: str, sep: str = ' ') → None[source]¶

Stores a request to remove flags from an environment variable

Parameters:

name – name of the environment variable
value – flags to be removed
sep – separator for the flags (default: " ")

remove_last_path(name: str, path: str | PurePath, separator: str = os.pathsep) → None[source]¶

Stores a request to remove last instance of path from a list of paths.

Parameters:

name – name of the environment variable
path – path to be removed
separator – separator for the paths (default: os.pathsep)

remove_path(name: str, path: str | PurePath, separator: str = os.pathsep) → None[source]¶

Stores a request to remove a path from a list of paths.

Parameters:

name – name of the environment variable
path – path to be removed
separator – separator for the paths (default: os.pathsep)

reversed() → EnvironmentModifications[source]¶

Returns the EnvironmentModifications object that will reverse self

Only creates reversals for additions to the environment, as reversing unset() and remove_path() modifications is impossible.

Reversible operations are set(), prepend_path(), append_path(), set_path(), and append_flags().

set(name: str, value: str, *, force: bool = False, raw: bool = False) → None[source]¶

Stores a request to set an environment variable.

Parameters:

name – name of the environment variable
value – value of the environment variable
force – if True, audit will not consider this modification a warning
raw – if True, format of value string is skipped

set_path(name: str, elements: List[str] | List[PurePath] | List[str | PurePath], separator: str = os.pathsep) → None[source]¶

Stores a request to set an environment variable to a list of paths, separated by a character defined in input.

Parameters:

name – name of the environment variable
elements – ordered list paths
separator – separator for the paths (default: os.pathsep)

shell_modifications(shell: str = DEFAULT_SHELL, explicit: bool = False, env: MutableMapping[str, str] | None = None) → str[source]¶: Return shell code to apply the modifications.

unset(name: str) → None[source]¶

Stores a request to unset an environment variable.

Parameters:: name – name of the environment variable

class spack.util.environment.NameModifier(name: str, *, separator: str = os.pathsep, trace: Trace | None = None)[source]¶

Bases: object

Base class for modifiers that act on the environment variable as a whole, and thus store just its name

execute(env: MutableMapping[str, str])[source]¶: Apply the modification to the mapping passed as input

name¶

separator¶

trace¶

class spack.util.environment.NamePathModifier(name: str, value: str | PurePath, *, separator: str = os.pathsep, trace: Trace | None = None)[source]¶

Bases: NameValueModifier

Base class for modifiers that modify the value of an environment variable that is a path.

class spack.util.environment.NameValueModifier(name: str, value: str, *, separator: str = os.pathsep, trace: Trace | None = None)[source]¶

Bases: object

Base class for modifiers that modify the value of an environment variable.

execute(env: MutableMapping[str, str])[source]¶: Apply the modification to the mapping passed as input

name¶

separator¶

trace¶

value¶

class spack.util.environment.PrependPath(name: str, value: str | PurePath, *, separator: str = os.pathsep, trace: Trace | None = None)[source]¶

Bases: NamePathModifier

execute(env: MutableMapping[str, str])[source]¶: Apply the modification to the mapping passed as input

class spack.util.environment.PruneDuplicatePaths(name: str, *, separator: str = os.pathsep, trace: Trace | None = None)[source]¶

Bases: NameModifier

execute(env: MutableMapping[str, str])[source]¶: Apply the modification to the mapping passed as input

name¶

separator¶

trace¶

class spack.util.environment.RemoveFirstPath(name: str, value: str | PurePath, *, separator: str = os.pathsep, trace: Trace | None = None)[source]¶

Bases: NamePathModifier

execute(env: MutableMapping[str, str])[source]¶: Apply the modification to the mapping passed as input

class spack.util.environment.RemoveFlagsEnv(name: str, value: str, *, separator: str = os.pathsep, trace: Trace | None = None)[source]¶

Bases: NameValueModifier

execute(env: MutableMapping[str, str])[source]¶: Apply the modification to the mapping passed as input

name¶

separator¶

trace¶

value¶

class spack.util.environment.RemoveLastPath(name: str, value: str | PurePath, *, separator: str = os.pathsep, trace: Trace | None = None)[source]¶

Bases: NamePathModifier

execute(env: MutableMapping[str, str])[source]¶: Apply the modification to the mapping passed as input

class spack.util.environment.RemovePath(name: str, value: str | PurePath, *, separator: str = os.pathsep, trace: Trace | None = None)[source]¶

Bases: NamePathModifier

execute(env: MutableMapping[str, str])[source]¶: Apply the modification to the mapping passed as input

spack.util.environment.SYSTEM_DIR_CASE_ENTRY¶: used in the compiler wrapper’s /usr/lib|/usr/lib64|...) case entry

class spack.util.environment.SetEnv(name: str, value: str, *, trace: Trace | None = None, force: bool = False, raw: bool = False)[source]¶

Bases: NameValueModifier

execute(env: MutableMapping[str, str])[source]¶: Apply the modification to the mapping passed as input

force¶

raw¶

class spack.util.environment.SetPath(name: str, value: List[str] | List[PurePath] | List[str | PurePath], *, separator: str = os.pathsep, trace: Trace | None = None)[source]¶

Bases: NameValueModifier

execute(env: MutableMapping[str, str])[source]¶: Apply the modification to the mapping passed as input

name¶

separator¶

trace¶

value¶

class spack.util.environment.Trace(*, filename: str, lineno: int, context: str)[source]¶

Bases: object

Trace information on a function call

context¶

filename¶

lineno¶

class spack.util.environment.UnsetEnv(name: str, *, separator: str = os.pathsep, trace: Trace | None = None)[source]¶

Bases: NameModifier

execute(env: MutableMapping[str, str])[source]¶: Apply the modification to the mapping passed as input

name¶

separator¶

trace¶

spack.util.environment.deprioritize_system_paths(paths: List[str]) → List[str][source]¶: Reorders input paths by putting system paths at the end of the list, otherwise preserving order.

spack.util.environment.dump_environment(path: str, environment: MutableMapping[str, str] | None = None)[source]¶

Dump an environment dictionary to a source-able file.

Parameters:

path – path of the file to write
environment – environment to be written. If None os.environ is used.

spack.util.environment.env_flag(name: str) → bool[source]¶: Given the name of an environment variable, returns True if the lowercase value is set to true or to 1, False otherwise.

spack.util.environment.environment_after_sourcing_files(*files: str | Tuple[str, ...], **kwargs) → Dict[str, str][source]¶

Returns a dictionary with the environment that one would have after sourcing the files passed as argument.

Parameters:

*files – each item can either be a string containing the path of the file to be sourced or a sequence, where the first element is the file to be sourced and the remaining are arguments to be passed to the command line

Keyword Arguments:

env (dict) – the initial environment (default: current environment)
shell (str) – the shell to use (default: /bin/bash or cmd.exe (Windows))
shell_options (str) – options passed to the shell (default: -c or /C (Windows))
source_command (str) – the command to run (default: source)
suppress_output (str) – redirect used to suppress output of command (default: &> /dev/null)
concatenate_on_success (str) – operator used to execute a command only when the previous command succeeds (default: &&)

spack.util.environment.filter_system_paths(paths: Iterable[str]) → List[str][source]¶: Returns a copy of the input where system paths are filtered out.

spack.util.environment.get_path(name: str) → List[str][source]¶: Given the name of an environment variable containing multiple paths separated by os.pathsep, returns a list of the paths.

spack.util.environment.inspect_path(root: str, inspections: MutableMapping[str, List[str]], exclude: Callable[[str], bool] | None = None) → EnvironmentModifications[source]¶

Inspects root to search for the subdirectories in inspections. Adds every path found to a list of prepend-path commands and returns it.

Parameters:

root – absolute path where to search for subdirectories
inspections – maps relative paths to a list of environment variables that will be modified if the path exists. The modifications are not performed immediately, but stored in a command object that is returned to client
exclude – optional callable. If present it must accept an absolute path and return True if it should be excluded from the inspection

Examples:

The following lines execute an inspection in /usr to search for /usr/include and /usr/lib64. If found we want to prepend /usr/include to CPATH and /usr/lib64 to MY_LIB64_PATH.

# Set up the dictionary containing the inspection
inspections = {
    "include": ["CPATH"],
    "lib64": ["MY_LIB64_PATH"]
}

# Get back the list of command needed to modify the environment
env = inspect_path("/usr", inspections)

# Eventually execute the commands
env.apply_modifications()

spack.util.environment.is_system_path(path: str) → bool[source]¶: Returns True if the argument is a system path, False otherwise.

spack.util.environment.path_put_first(var_name: str, directories: List[str])[source]¶: Puts the provided directories first in the path, adding them if they’re not already there.

spack.util.environment.path_set(var_name: str, directories: List[str])[source]¶: Sets the variable passed as input to the os.pathsep joined list of directories.

spack.util.environment.pickle_environment(path: str, environment: Dict[str, str] | None = None)[source]¶: Pickle an environment dictionary to a file.

spack.util.environment.preserve_environment(*variables: str)[source]¶

Ensures that the value of the environment variables passed as arguments is the same before entering to the context manager and after exiting it.

Variables that are unset before entering the context manager will be explicitly unset on exit.

Parameters:: variables – list of environment variables to be preserved

spack.util.environment.prune_duplicate_paths(paths: List[str]) → List[str][source]¶: Returns the input list with duplicates removed, otherwise preserving order.

spack.util.environment.sanitize(environment: MutableMapping[str, str], exclude: List[str], include: List[str]) → Dict[str, str][source]¶

Returns a copy of the input dictionary where all the keys that match an excluded pattern and don’t match an included pattern are removed.

Parameters:

environment (dict) – input dictionary
exclude (list) – literals or regex patterns to be excluded
include (list) – literals or regex patterns to be included

spack.util.environment.set_env(**kwargs)[source]¶: Temporarily sets and restores environment variables. Variables can be set as keyword arguments to this function.

Note

If the goal is to set environment variables for a subprocess, it is strongly recommended to use the extra_env argument of spack.util.executable.Executable.__call__() instead of this function.

This function is intended to modify the current process’s environment (which is an unsafe operation in general).

spack.util.environment.validate(env: EnvironmentModifications, errstream: Callable[[str], None])[source]¶

Validates the environment modifications to check for the presence of suspicious patterns. Prompts a warning for everything that was found.

Current checks: - set or unset variables after other changes on the same variable

Parameters:

env – list of environment modifications
errstream – callable to log error messages

spack.util.executable module¶

class spack.util.executable.Executable(name: str | Path)[source]¶

Bases: object

Represent an executable file that can be run as a subprocess.

This class provides a simple interface for running executables with custom arguments and environment variables. It supports setting default arguments and environment modifications, copying instances, and running commands with various options for input/output/error handling.

Example usage:

ls = Executable("ls")
ls.add_default_arg("-l")
ls.add_default_env("LC_ALL", "C")
output = ls("-a", output=str)  # Run 'ls -l -a' and capture output as string

Runs this executable in a subprocess.

Parameters:

*args – command-line arguments to the executable to run
fail_on_error – if True, raises an exception if the subprocess returns an error The return code is available as returncode
ignore_errors – a sequence of error codes to ignore. If these codes are returned, this process will not raise an exception, even if fail_on_error is set to True
ignore_quotes – if False, warn users that quotes are not needed, as Spack does not use a shell. If None, use ignore_quotes.
timeout – the number of seconds to wait before killing the child process
env – the environment with which to run the executable
extra_env – extra items to add to the environment (neither requires nor precludes env)
input – where to read stdin from
output – where to send stdout
error – where to send stderr
_dump_env – dict to be set to the environment actually used (envisaged for testing purposes only)

Accepted values for input, output, and error:

Python streams: open Python file objects or os.devnull
str: the Python string type. If you set these to str, output and error will be written to pipes and returned as a string. If both output and error are set to str, then one string is returned containing output concatenated with error. Not valid for input.
str.split: the split method of the Python string type. Behaves the same as str, except that value is also written to stdout or stderr.

For output and error it’s also accepted to pass a string with a filename, which will be automatically opened for writing.

By default, the subprocess inherits the parent’s file descriptors.

add_default_arg(*args: str) → None[source]¶: Add default argument(s) to the command.

add_default_env(key: str, value: str) → None[source]¶

Set an environment variable when the command is run.

Parameters:

key – The environment variable to set
value – The value to set it to

add_default_envmod(envmod: EnvironmentModifications) → None[source]¶: Set an spack.util.environment.EnvironmentModifications to use when the command is run.

property command: str¶: Returns the entire command-line string

copy() → Executable[source]¶: Return a copy of this Executable.

ignore_quotes: bool¶: Whether to warn users that quotes are not needed, as Spack does not use a shell.

property name: str¶: Returns the executable name

property path: str¶: Returns the executable path

returncode: int¶: Return code of the last executed command.

with_default_args(*args: str) → Executable[source]¶: Same as add_default_arg, but returns a copy of the executable.

exception spack.util.executable.ProcessError(message: str, long_message: str | None = None)[source]¶

Bases: SpackError

Raised when Executable exits with an error code.

spack.util.executable.which(*args: str, path: str | List[str] | None = None, required: Literal[True]) → Executable[source]¶

spack.util.executable.which(*args: str, path: str | List[str] | None = None, required: bool = False) → Executable | None

Finds an executable in the path like command-line which.

If given multiple executables, returns the first one that is found. If no executables are found, returns None.

Parameters:

*args – one or more executables to search for
path – the path to search. Defaults to PATH
required – if set to True, raise an error if executable not found

Returns:

The first executable that is found in the path or None if not found.

spack.util.executable.which_string(*args: str, path: str | List[str] | None = None, required: Literal[True]) → str[source]¶
spack.util.executable.which_string(*args: str, path: str | List[str] | None = None, required: bool = False) → str | None: Like which(), but returns a string instead of an Executable.

spack.util.file_cache module¶

exception spack.util.file_cache.CacheError(message: str, long_message: str | None = None)[source]¶: Bases: SpackError

class spack.util.file_cache.FileCache(root: str | Path, timeout=120)[source]¶

Bases: object

This class manages cached data in the filesystem.

Cache files are fetched and stored by unique keys. Keys can be relative paths, so that there can be some hierarchy in the cache.
The FileCache handles locking cache files for reading and writing, so client code need not manage locks for cache entries.

cache_path(key: str | Path)[source]¶: Path to the file in the cache for a particular key.

destroy()[source]¶: Remove all files under the cache root.

read_transaction(key: str | Path) → Iterator[IO[str] | None][source]¶

Get a read transaction on a file cache item.

Returns a context manager that yields an open file object for reading, or None if the cache file does not exist. You can use it like this:

with file_cache_object.read_transaction(key) as cache_file:
    if cache_file is not None:
        cache_file.read()

remove(key: str | Path)[source]¶

write_transaction(key: str | Path) → Iterator[Tuple[IO[str] | None, IO[str]]][source]¶

Get a write transaction on a file cache item.

Returns a context manager that yields (old_file, new_file) where old_file is the existing cache file (or None), and new_file is a writable temporary file. Once the context manager exits cleanly, moves the temporary file into place atomically.

class spack.util.file_cache.ReadContextManager(path: str | Path)[source]¶: Bases: object

class spack.util.file_cache.WriteContextManager(path: str | Path)[source]¶: Bases: object

spack.util.file_permissions module¶

exception spack.util.file_permissions.InvalidPermissionsError(message: str, long_message: str | None = None)[source]¶

Bases: SpackError

Error class for invalid permission setters

spack.util.file_permissions.set_permissions(path, perms, group=None)[source]¶

spack.util.file_permissions.set_permissions_by_spec(path, spec)[source]¶

spack.util.filesystem module¶

class spack.util.filesystem.BaseDirectoryVisitor[source]¶

Bases: object

Base class and interface for visit_directory_tree().

after_visit_dir(root: str, rel_path: str, depth: int) → None[source]¶

Called after recursion into rel_path finished. This function is not called when rel_path was not recursed into.

Parameters:

root – root directory
rel_path – relative path to current directory from root
depth – depth of current directory from the root directory

after_visit_symlinked_dir(root: str, rel_path: str, depth: int) → None[source]¶

Called after recursion into rel_path finished. This function is not called when rel_path was not recursed into.

Parameters:

root – root directory
rel_path – relative path to current symlink from root
depth – depth of current symlink from the root directory

before_visit_dir(root: str, rel_path: str, depth: int) → bool[source]¶

Return True from this function to recurse into the directory at os.path.join(root, rel_path). Return False in order not to recurse further.

Parameters:

root – root directory
rel_path – relative path to current directory from root
depth – depth of current directory from the root directory

Returns:

True when the directory should be recursed into. False when not

Return type:

bool

before_visit_symlinked_dir(root: str, rel_path: str, depth: int) → bool[source]¶

Return True to recurse into the symlinked directory and False in order not to. Note: rel_path is the path to the symlink itself. Following symlinked directories blindly can cause infinite recursion due to cycles.

Parameters:

root – root directory
rel_path – relative path to current symlink from root
depth – depth of current symlink from the root directory

Returns:

True when the directory should be recursed into. False when not

Return type:

bool

visit_file(root: str, rel_path: str, depth: int) → None[source]¶

Handle the non-symlink file at os.path.join(root, rel_path)

Parameters:

root – root directory
rel_path – relative path to current file from root
depth (int) – depth of current file from the root directory

visit_symlinked_file(root: str, rel_path: str, depth) → None[source]¶

Handle the symlink to a file at os.path.join(root, rel_path). Note: rel_path is the location of the symlink, not to what it is pointing to. The symlink may be dangling.

Parameters:

root – root directory
rel_path – relative path to current symlink from root
depth – depth of current symlink from the root directory

class spack.util.filesystem.FileFilter(*filenames)[source]¶

Bases: object

Convenience class for repeatedly applying filter_file() to one or more files.

This class allows you to specify a set of filenames and then call filter() multiple times to perform search-and-replace operations using Python regular expressions, similar to sed.

Example usage:

foo_c = FileFilter("foo.c")
foo_c.filter(r"#define FOO", "#define BAR")
foo_c.filter(r"old_func", "new_func")

filter(regex: str, repl: str | Callable[[Match], str], string: bool = False, backup: bool = False, ignore_absent: bool = False, start_at: str | None = None, stop_at: str | None = None) → None[source]¶

class spack.util.filesystem.FileList(files: str | Iterable[str])[source]¶

Bases: Sequence

Sequence of absolute paths to files.

Provides a few convenience methods to manipulate file paths.

property basenames: List[str]¶

Stable de-duplication of the base-names in the list

>>> l = LibraryList(["/dir1/liba.a", "/dir2/libb.a", "/dir3/liba.a"])
>>> l.basenames
["liba.a", "libb.a"]
>>> h = HeaderList(["/dir1/a.h", "/dir2/b.h", "/dir3/a.h"])
>>> h.basenames
["a.h", "b.h"]

Returns:: A list of base-names

property directories: List[str]¶

Stable de-duplication of the directories where the files reside.

>>> l = LibraryList(["/dir1/liba.a", "/dir2/libb.a", "/dir1/libc.a"])
>>> l.directories
["/dir1", "/dir2"]
>>> h = HeaderList(["/dir1/a.h", "/dir1/b.h", "/dir2/c.h"])
>>> h.directories
["/dir1", "/dir2"]

Returns:: A list of directories

joined(separator: str = ' ') → str[source]¶

class spack.util.filesystem.HeaderList(files)[source]¶

Bases: FileList

Sequence of absolute paths to headers.

Provides a few convenience methods to manipulate header paths and get commonly used compiler flags or names.

add_macro(macro: str) → None[source]¶

Add a macro definition

Parameters:: macro – The macro to add

property cpp_flags: str¶

Include flags + macro definitions

>>> h = HeaderList(["/dir1/a.h", "/dir1/b.h", "/dir2/c.h"])
>>> h.cpp_flags
"-I/dir1 -I/dir2"
>>> h.add_macro("-DBOOST_DYN_LINK")
>>> h.cpp_flags
"-I/dir1 -I/dir2 -DBOOST_DYN_LINK"

Returns:: A joined list of include flags and macro definitions

property directories: List[str]¶: Directories to be searched for header files.

property headers: List[str]¶

Stable de-duplication of the headers.

Returns:: A list of header files

property include_flags: str¶

Include flags

>>> h = HeaderList(["/dir1/a.h", "/dir1/b.h", "/dir2/c.h"])
>>> h.include_flags
"-I/dir1 -I/dir2"

Returns:: A joined list of include flags

include_regex¶

property macro_definitions: str¶

Macro definitions

>>> h = HeaderList(["/dir1/a.h", "/dir1/b.h", "/dir2/c.h"])
>>> h.add_macro("-DBOOST_LIB_NAME=boost_regex")
>>> h.add_macro("-DBOOST_DYN_LINK")
>>> h.macro_definitions
"-DBOOST_LIB_NAME=boost_regex -DBOOST_DYN_LINK"

Returns:: A joined list of macro definitions

property names: List[str]¶

Stable de-duplication of header names in the list without extensions

>>> h = HeaderList(["/dir1/a.h", "/dir2/b.h", "/dir3/a.h"])
>>> h.names
["a", "b"]

Returns:: A list of files without extensions

class spack.util.filesystem.LibraryList(files: str | Iterable[str])[source]¶

Bases: FileList

Sequence of absolute paths to libraries

Provides a few convenience methods to manipulate library paths and get commonly used compiler flags or names

property ld_flags: str¶

Search flags + link flags

>>> l = LibraryList(["/dir1/liba.a", "/dir2/libb.a", "/dir1/liba.so"])
>>> l.ld_flags
"-L/dir1 -L/dir2 -la -lb"

Returns:: A joined list of search flags and link flags

property libraries: List[str]¶

Stable de-duplication of library files.

Returns:: A list of library files

property link_flags: str¶

Link flags for the libraries

>>> l = LibraryList(["/dir1/liba.a", "/dir2/libb.a", "/dir1/liba.so"])
>>> l.link_flags
"-la -lb"

Returns:: A joined list of link flags

property names: List[str]¶

Stable de-duplication of library names in the list

>>> l = LibraryList(["/dir1/liba.a", "/dir2/libb.a", "/dir3/liba.so"])
>>> l.names
["a", "b"]

Returns:: A list of library names

property search_flags: str¶

Search flags for the libraries

>>> l = LibraryList(["/dir1/liba.a", "/dir2/libb.a", "/dir1/liba.so"])
>>> l.search_flags
"-L/dir1 -L/dir2"

Returns:: A joined list of search flags

spack.util.filesystem.ancestor(dir, n=1)[source]¶: Get the nth ancestor of a directory.

spack.util.filesystem.can_access(file_name)[source]¶: True if the current process has read and write access to the file.

spack.util.filesystem.change_sed_delimiter(old_delim: str, new_delim: str, *filenames: str) → None[source]¶

Find all sed search/replace commands and change the delimiter.

e.g., if the file contains seds that look like 's///', you can call change_sed_delimiter('/', '@', file) to change the delimiter to '@'.

Note that this routine will fail if the delimiter is ' or ". Handling those is left for future work.

Parameters:

old_delim – The delimiter to search for
new_delim – The delimiter to replace with
*filenames – One or more files to search and replace

spack.util.filesystem.chgrp(path, group, follow_symlinks=True)[source]¶: Implement the bash chgrp function on a single path

spack.util.filesystem.chmod_x(entry, perms)[source]¶: Implements chmod, treating all executable bits as set using the chmod utility’s +X option.

spack.util.filesystem.copy(src: str, dest: str, _permissions: bool = False) → None[source]¶

Copy the file(s) src to the file or directory dest.

If dest specifies a directory, the file will be copied into dest using the base filename from src.

src may contain glob characters.

Parameters:

src – the file(s) to copy
dest – the destination file or directory
_permissions – for internal use only

Raises:

OSError – if src does not match any files or directories
ValueError – if src matches multiple files but dest is not a directory

spack.util.filesystem.copy_mode(src, dest)[source]¶: Set the mode of dest to that of src unless it is a link.

spack.util.filesystem.copy_tree(src: str, dest: str, symlinks: bool = True, ignore: Callable[[str], bool] | None = None, _permissions: bool = False)[source]¶

Recursively copy an entire directory tree rooted at src.

If the destination directory dest does not already exist, it will be created as well as missing parent directories.

src may contain glob characters.

If symlinks is true, symbolic links in the source tree are represented as symbolic links in the new tree and the metadata of the original links will be copied as far as the platform allows; if false, the contents and metadata of the linked files are copied to the new tree.

If ignore is set, then each path relative to src will be passed to this function; the function returns whether that path should be skipped.

Parameters:

src – the directory to copy
dest – the destination directory
symlinks – whether or not to preserve symlinks
ignore – function indicating which files to ignore
_permissions – for internal use only

Raises:

OSError – if src does not match any files or directories
ValueError – if src is a parent directory of dest

spack.util.filesystem.filter_file(regex: str, repl: str | Callable[[Match], str], *filenames: str, string: bool = False, backup: bool = False, ignore_absent: bool = False, start_at: str | None = None, stop_at: str | None = None, encoding: str | None = 'utf-8') → None[source]¶

Like sed, but uses Python regular expressions.

Filters every line of each file through regex and replaces the file with a filtered version. Preserves mode of filtered files.

As with re.sub(), repl can be either a string or a callable. If it is a callable, it is passed the match object and should return a suitable replacement string. If it is a string, it can contain \1, \2, etc. to represent back-substitution as sed would allow.

Parameters:

regex – The regular expression to search for
repl – The string to replace matches with
*filenames – One or more files to search and replace string: Treat regex as a plain string. Default it False backup: Make backup file(s) suffixed with ~. Default is False
ignore_absent – Ignore any files that don’t exist. Default is False
start_at – Marker used to start applying the replacements. If a text line matches this marker filtering is started at the next line. All contents before the marker and the marker itself are copied verbatim. Default is to start filtering from the first line of the file.
stop_at – Marker used to stop scanning the file further. If a text line matches this marker filtering is stopped and the rest of the file is copied verbatim. Default is to filter until the end of the file.
encoding – The encoding to use when reading and writing the files. Default is None, which uses the system’s default encoding.

Finds all files matching the patterns from files starting from root. This function returns a deterministic result for the same input and directory structure when run multiple times. Symlinked directories are followed, and unique directories are searched only once. Each matching file is returned only once at lowest depth in case multiple paths exist due to symlinked directories.

Accepts any glob characters accepted by fnmatch.fnmatch():

Pattern	Meaning
`*`	matches one or more characters
`?`	matches any single character
`[seq]`	matches any character in `seq`
`[!seq]`	matches any character not in `seq`

Examples:

>>> find("/usr", "*.txt", recursive=True, max_depth=2)

finds all files with the extension .txt in the directory /usr and subdirectories up to depth 2.

>>> find(["/usr", "/var"], ["*.txt", "*.log"], recursive=True)

finds all files with the extension .txt or .log in the directories /usr and /var at any depth.

>>> find("/usr", "GL/*.h", recursive=True)

finds all header files in a directory GL at any depth in the directory /usr.

Parameters:

root – One or more root directories to start searching from
files – One or more filename patterns to search for
recursive – if False search only root, if True descends from roots. Defaults to True.
max_depth – if set, don’t search below this depth. Cannot be set if recursive is False

Returns a list of absolute, matching file paths.

spack.util.filesystem.find_all_headers(root: str) → HeaderList[source]¶

Convenience function that returns the list of all headers found in the directory passed as argument.

Parameters:: root – directory where to look recursively for header files
Returns:: List of all headers found in root and subdirectories.

spack.util.filesystem.find_first(root: str, files: Iterable[str] | str, bfs_depth: int = 2) → str | None[source]¶

Find the first file matching a pattern.

The following

$ find /usr -name 'abc*' -o -name 'def*' -quit

is equivalent to:

>>> find_first("/usr", ["abc*", "def*"])

Any glob pattern supported by fnmatch can be used.

The search order of this method is breadth-first over directories, until depth bfs_depth, after which depth-first search is used.

Parameters:

root – The root directory to start searching from
files – File pattern(s) to search for
bfs_depth – (advanced) parameter that specifies at which depth to switch to depth-first search.

Returns:

The matching file or None when no file is found.

spack.util.filesystem.find_headers(headers: str | List[str], root: str, recursive: bool = False) → HeaderList[source]¶

Returns an iterable object containing a list of full paths to headers if found.

Accepts any glob characters accepted by fnmatch.fnmatch():

Pattern	Meaning
`*`	matches one or more characters
`?`	matches any single character
`[seq]`	matches any character in `seq`
`[!seq]`	matches any character not in `seq`

Parameters:

headers – Header name(s) to search for
root – The root directory to start searching from
recursive – if False search only root folder, if True descends top-down from the root. Defaults to False.

Returns:

The headers that have been found

spack.util.filesystem.find_libraries(libraries: str | List[str], root: str, shared: bool = True, recursive: bool = False, runtime: bool = True, max_depth: int | None = None) → LibraryList[source]¶

Returns an iterable of full paths to libraries found in a root dir.

Accepts any glob characters accepted by fnmatch.fnmatch():

Pattern	Meaning
`*`	matches one or more characters
`?`	matches any single character
`[seq]`	matches any character in `seq`
`[!seq]`	matches any character not in `seq`

Parameters:

libraries – Library name(s) to search for
root – The root directory to start searching from
shared – if True searches for shared libraries, otherwise for static. Defaults to True.
recursive – if False search only root folder, if True descends top-down from the root. Defaults to False.
max_depth – if set, don’t search below this depth. Cannot be set if recursive is False
runtime – Windows only option, no-op elsewhere. If True, search for runtime shared libs (.DLL), otherwise, search for .Lib files. If shared is False, this has no meaning. Defaults to True.

Returns:

The libraries that have been found

spack.util.filesystem.find_system_libraries(libraries: str | List[str], shared: bool = True) → LibraryList[source]¶

Searches the usual system library locations for libraries.

Search order is as follows:

/lib64
/lib
/usr/lib64
/usr/lib
/usr/local/lib64
/usr/local/lib

Accepts any glob characters accepted by fnmatch.fnmatch():

Pattern	Meaning
`*`	matches one or more characters
`?`	matches any single character
`[seq]`	matches any character in `seq`
`[!seq]`	matches any character not in `seq`

Parameters:

libraries – Library name(s) to search for
shared – if True searches for shared libraries, otherwise for static. Defaults to True.

Returns:

The libraries that have been found

spack.util.filesystem.fix_darwin_install_name(path: str) → None[source]¶

Fix install name of dynamic libraries on Darwin to have full path.

There are two parts of this task:

Use install_name -id ... to change install name of a single lib
Use install_name -change ... to change the cross linking between libs. The function assumes that all libraries are in one folder and currently won’t follow subfolders.

Parameters:: path – directory in which .dylib files are located

spack.util.filesystem.force_remove(*paths: str) → None[source]¶: Remove files without printing errors. Like rm -f, does NOT remove directories.

spack.util.filesystem.force_symlink(src: str, dest: str) → None[source]¶: Create a symlink at dest pointing to src. Similar to ln -sf.

spack.util.filesystem.getuid()¶: Return the current process’s user id.

spack.util.filesystem.install(src: str, dest: str) → None[source]¶

Install the file(s) src to the file or directory dest.

Same as copy() with the addition of setting proper permissions on the installed file.

Parameters:

src – the file(s) to install
dest – the destination file or directory

Raises:

OSError – if src does not match any files or directories
ValueError – if src matches multiple files but dest is not a directory

spack.util.filesystem.install_tree(src: str, dest: str, symlinks: bool = True, ignore: Callable[[str], bool] | None = None)[source]¶

Recursively install an entire directory tree rooted at src.

Same as copy_tree() with the addition of setting proper permissions on the installed files and directories.

Parameters:

src – the directory to install
dest – the destination directory
symlinks – whether or not to preserve symlinks
ignore – function indicating which files to ignore

Raises:

OSError – if src does not match any files or directories
ValueError – if src is a parent directory of dest

spack.util.filesystem.is_exe(path) → bool[source]¶: Returns True iff the specified path exists, is a regular file, and has executable permissions for the current process.

spack.util.filesystem.join_path(prefix, *args) → str[source]¶: Alias for os.path.join()

spack.util.filesystem.keep_modification_time(*filenames: str) → Generator[None, None, None][source]¶

Context manager to keep the modification timestamps of the input files. Tolerates and has no effect on non-existent files and files that are deleted by the nested code.

Example:

with keep_modification_time("file1.txt", "file2.txt"):
    # do something that modifies file1.txt and file2.txt

Parameters:: *filenames – one or more files that must have their modification timestamps unchanged

spack.util.filesystem.library_extensions¶: This generates the library filenames that may appear on any OS.

spack.util.filesystem.mkdirp(*paths: str, mode: int | None = None, group: str | int | None = None, default_perms: str | None = None)[source]¶

Creates a directory, as well as parent directories if needed.

Parameters:

paths – paths to create with mkdirp
mode – optional permissions to set on the created directory – use OS default if not provided
group – optional group for permissions of final created directory – use OS default if not provided. Only used if world write permissions are not set
default_perms – one of "parents" or "args". The default permissions that are set for directories that are not themselves an argument for mkdirp. "parents" means intermediate directories get the permissions of their direct parent directory, "args" means intermediate get the same permissions specified in the arguments to mkdirp – default value is "args"

spack.util.filesystem.partition_path(path, entry=None)[source]¶

Split the prefixes of the path at the first occurrence of entry and return a 3-tuple containing a list of the prefixes before the entry, a string of the prefix ending with the entry, and a list of the prefixes after the entry.

If the entry is not a node in the path, the result will be the prefix list followed by an empty string and an empty list.

spack.util.filesystem.prefixes(path)[source]¶

Returns a list containing the path and its ancestors, top-to-bottom.

The list for an absolute path will not include an os.sep entry. For example, assuming os.sep is /, given path /ab/cd/efg the resulting paths will be, in order: /ab, /ab/cd, and /ab/cd/efg

The list for a relative path starting ./ will not include .. For example, path ./hi/jkl/mn results in a list with the following paths, in order: ./hi, ./hi/jkl, and ./hi/jkl/mn.

On Windows, paths will be normalized to use / and / will always be used as the separator instead of os.sep.

Parameters:: path (str) – the string used to derive ancestor paths
Returns:: A list containing ancestor paths in order and ending with the path

spack.util.filesystem.remove_dead_links(root)[source]¶

Recursively removes any dead link that is present in root.

Parameters:: root (str) – path where to search for dead links

spack.util.filesystem.remove_directory_contents(dir)[source]¶: Remove all contents of a directory.

spack.util.filesystem.remove_if_dead_link(path)[source]¶

Removes the argument if it is a dead link.

Parameters:: path (str) – The potential dead link

spack.util.filesystem.remove_linked_tree(path: str) → None[source]¶

Removes a directory and its contents.

If the directory is a symlink, follows the link and removes the real directory before removing the link.

This method will force-delete files on Windows

Parameters:: path – Directory to be removed

spack.util.filesystem.rename(src, dst, *, src_dir_fd=None, dst_dir_fd=None)¶

Rename a file or directory.

If either src_dir_fd or dst_dir_fd is not None, it should be a file descriptor open to a directory, and the respective path string (src or dst) should be relative; the path will then be relative to that directory. src_dir_fd and dst_dir_fd, may not be implemented on your platform. If they are unavailable, using them will raise a NotImplementedError.

spack.util.filesystem.set_executable(path)[source]¶: Set the executable bit on a file or directory.

spack.util.filesystem.set_install_permissions(path)[source]¶: Set appropriate permissions on the installed file.

spack.util.filesystem.touch(path)[source]¶: Creates an empty file at the specified path.

spack.util.filesystem.touchp(path)[source]¶: Like touch, but creates any parent directories needed for the file.

spack.util.filesystem.traverse_tree(source_root: str, dest_root: str, rel_path: str = '', *, order: str = 'pre', ignore: Callable[[str], bool] | None = None, follow_nonexisting: bool = True, follow_links: bool = False)[source]¶

Traverse two filesystem trees simultaneously.

Walks the LinkTree directory in pre or post order. Yields each file in the source directory with a matching path from the dest directory, along with whether the file is a directory. e.g., for this tree:

root/
  a/
    file1
    file2
  b/
    file3

When called on dest, this yields:

("root",         "dest")
("root/a",       "dest/a")
("root/a/file1", "dest/a/file1")
("root/a/file2", "dest/a/file2")
("root/b",       "dest/b")
("root/b/file3", "dest/b/file3")

Keyword Arguments:

order (str) – Whether to do pre- or post-order traversal. Accepted values are "pre" and "post"
ignore (Callable) – function indicating which files to ignore. This will also ignore symlinks if they point to an ignored file (regardless of whether the symlink is explicitly ignored); note this only supports one layer of indirection (i.e. if you have x -> y -> z, and z is ignored but x/y are not, then y would be ignored but not x). To avoid this, make sure the ignore function also ignores the symlink paths too.
follow_nonexisting (bool) – Whether to descend into directories in src that do not exit in dest. Default is True
follow_links (bool) – Whether to descend into symlinks in src

spack.util.filesystem.unset_executable_mode(path)[source]¶

spack.util.filesystem.visit_directory_tree(root: str, visitor: BaseDirectoryVisitor, rel_path: str = '', depth: int = 0)[source]¶

Recurses the directory root depth-first through a visitor pattern using the interface from BaseDirectoryVisitor

Parameters:

root – path of directory to recurse into
visitor – what visitor to use
rel_path – current relative path from the root
depth – current depth from the root

spack.util.filesystem.working_dir(dirname: str, *, create: bool = False)[source]¶

Context manager to change the current working directory to dirname.

Parameters:

dirname – the directory to change to
create – if True, create the directory if it does not exist

Example usage:

with working_dir("/path/to/dir"):
    # do something in /path/to/dir
    pass

spack.util.format module¶

spack.util.format.get_version_lines(version_hashes_dict: dict) → str[source]¶

Renders out a set of versions like those found in a package’s package.py file for a given set of versions and hashes.

Parameters:: version_hashes_dict – A dictionary of the form: version -> checksum.

Returns: Rendered version lines.

spack.util.gcs module¶

This file contains the definition of the GCS Blob storage Class used to integrate GCS Blob storage with spack buildcache.

class spack.util.gcs.GCSBlob(url, client=None)[source]¶

Bases: object

GCS Blob object

Wraps some blob methods for spack functionality

delete_blob()[source]¶

exists()[source]¶

get()[source]¶

get_blob_byte_stream()[source]¶

get_blob_headers()[source]¶

upload_to_blob(local_file_path)[source]¶

class spack.util.gcs.GCSBucket(url, client=None)[source]¶

Bases: object

GCS Bucket Object Create a wrapper object for a GCS Bucket. Provides methods to wrap spack related tasks, such as destroy.

blob(blob_path)[source]¶

create()[source]¶

destroy(recursive=False, **kwargs)[source]¶

Bucket destruction method

Deletes all blobs within the bucket, and then deletes the bucket itself.

Uses GCS Batch operations to bundle several delete operations together.

exists()[source]¶

get_all_blobs(recursive: bool = True, relative: bool = True) → List[str][source]¶

Get a list of all blobs

Returns: a list of all blobs within this bucket.

Parameters:: relative – If true (default), print blob paths relative to ‘build_cache’ directory. If false, print absolute blob paths (useful for destruction of bucket)

get_blob(blob_path)[source]¶

class spack.util.gcs.GCSHandler[source]¶

Bases: BaseHandler

gs_open(req)[source]¶

spack.util.gcs.gcs_client()[source]¶: Create a GCS client Creates an authenticated GCS client to access GCS buckets and blobs

spack.util.gcs.gcs_open(req, *args, **kwargs)[source]¶: Open a reader stream to a blob object on GCS

spack.util.git module¶

Single util module where Spack should get a git executable.

class spack.util.git.GitExecutable(name=None)[source]¶

Bases: Executable

Specialized executable that encodes the git version for optimized option selection

property version¶

class spack.util.git.VersionConditionalOption(key, value=None, min_version=(0, 0, 0), max_version=(99, 99, 99))[source]¶: Bases: object

spack.util.git.extract_git_version_str(git_exe: Executable) → str[source]¶

spack.util.git.get_commit_sha(path: str, ref: str) → str | None[source]¶: Get a commit sha for an arbitrary ref using ls-remote

spack.util.git.get_modified_files(from_ref: str = 'HEAD~1', to_ref: str = 'HEAD', git_exe: Executable | None = None) → List[str][source]¶

Get a list of files modified between from_ref and to_ref :param from_ref: oldest git ref, defaults to HEAD~1 :type from_ref: str :param to_ref: newer git ref, defaults to HEAD :type to_ref: str

Returns: list of file paths

spack.util.git.git(required: Literal[True]) → GitExecutable[source]¶

spack.util.git.git(required: bool = False) → GitExecutable | None

Get a git executable.

The returned executable automatically unsets GIT_EXTERNAL_DIFF and GIT_DIFF_OPTS environment variables that can interfere with spack git diff operations.

Parameters:: required (bool) – if True, raises CommandNotFoundError when git is not found

Returns: GitExecutable, or None if git is not found and required is False

spack.util.git.git_checkout(ref: str | None = None, sparse_paths: List[str] = [], debug: bool = False, dest: str | None = None, git_exe: GitExecutable | None = None)[source]¶: A generic method for running git checkout that integrates sparse-checkout Several methods in this module explicitly delay checkout so sparse-checkout can be called. It is intended to be used with git clone --no-checkout or git init && git fetch. There is minimal impact to performance since the initial clone operation filters blobs and has to download a minimal subset of git data.

spack.util.git.git_clone(url: str, ref: str | None = None, full_repo: bool = False, depth: int | None = None, debug: bool = False, dest: str | None = None, git_exe: GitExecutable | None = None)[source]¶: A git clone that prefers deferring expensive blob fetching for modern git installations This is our fallback method for capturing more git data than the init && fetch model. It is still optimized to capture a minimal set of ./.git data and expects to be paired with a call to git checkout to fully download the source code.

spack.util.git.git_init_fetch(url, ref, depth=None, debug=False, dest=None, git_exe=None)[source]¶

Utilize git init and then git fetch for a minimal clone of a single git ref This method runs git init, repo add, fetch to get a minimal set of source data. Profiling has shown this method can be 10-20% less storage than purely using sparse-checkout, and is even smaller than git clone –depth 1. This makes it the preferred method for single commit checkouts and source mirror population.

There is a trade off since less git data means less flexibility with additional git operations. Technically adding the remote is not necessary, but we do it since there are test cases where we may want to fetch additional data.

Checkout is explicitly deferred to a second method so we can intercept and add sparse-checkout options uniformly whether we use git clone or init fetch

spack.util.git.init_git_repo(repository: str, remote: str = 'origin', git_exe: Executable | None = None)[source]¶: Initialize a new Git repository and configure it with a remote.

spack.util.git.is_git_commit_sha(string: str) → bool[source]¶

spack.util.git.protocol_supports_shallow_clone(url)[source]¶: Shallow clone operations (--depth #) are not supported by the basic HTTP protocol or by no-protocol file specifications. Use (e.g.) https:// or file:// instead.

spack.util.git.pull_checkout_branch(branch: str, remote: str = 'origin', depth: int | None = None, git_exe: Executable | None = None)[source]¶: Fetch and checkout branch, then rebase with remote tracking branch.

spack.util.git.pull_checkout_commit(commit: str, remote: str | None = None, depth: int | None = None, git_exe: Executable | None = None)[source]¶: Checkout the specified commit (fetched if necessary).

spack.util.git.pull_checkout_tag(tag: str, remote: str = 'origin', depth: int | None = None, git_exe: Executable | None = None)[source]¶: Fetch tags with specified depth and checkout the given tag.

spack.util.gpg module¶

spack.util.gpg.CLEARSIGN_FILE_REGEX¶: Regular expression to pull spec contents out of clearsigned signature file.

spack.util.gpg.GNUPGHOME¶: GNUPGHOME environment variable in the context of this Python module

spack.util.gpg.GPG: Gpg | None¶: Executable instance for “gpg”, initialized lazily

spack.util.gpg.GPGCONF: Executable | None¶: Executable instance for “gpgconf”, initialized lazily

class spack.util.gpg.Gpg(gnupghome: str | None = None)[source]¶

Bases: object

Wrapper for GPG

property conf: Executable | None¶

export_keys(keyfile: str, keys: List[GpgKey], ktype: GpgKeyType = GpgKeyType.PUBLIC)[source]¶

Export public keys to a location passed as argument.

Parameters:

keyfile – where to export the keys
keys – keys to be exported
secret – whether to export secret keys or not

property gpg¶

keys(*fprs, ktype: GpgKeyType = GpgKeyType.PUBLIC) → List[GpgKey][source]¶

list_keyfile(keyfile: str, ktype: GpgKeyType | List[GpgKeyType] = GpgKeyType.PUBLIC) → List[GpgKey][source]¶: List keys in a keyfile

list_keys(*fprs, ktype: GpgKeyType = GpgKeyType.PUBLIC, fmt: str = '') → str | List[GpgKey][source]¶

List known keys.

Parameters:

fprs – list of key fingerprints
ktype – Type of dey to list (default: PUBLIC)
fmt – format to print/return keys (default: None) default (aka “”) -> return default output from gpg GpgKey format string-> See GpgKey __format__

Sign a file with a key.

Parameters:

blob – file to be signed
output – output file (default: f”{blob}.sig”)
key – key to be used to sign (default: first secret key in keyring)
armor – ascii armored output
clearsign – if True wraps the document in an ASCII-armored signature, if False creates a detached signature

property socket_dir: Path | None¶

trust(keyfile: str, *, fprs: List[str] | None = None, ownertrust: GpgKeyTrust = GpgKeyTrust.ULTIMATE, yes_to_all: bool = False)[source]¶

Import a key from a file and trust it.

The keyfile may contain public keys, secret keys (which embed public key material), or both.

Parameters:

keyfile – file with the public or secret key(s)
fprs – list of fingerprints to trust, if provided, then yes_to_all is ignored
ownertrust – level of trust to assign to the key(s)
yes_to_all – trust all keys in the file if True, otherwise ask for each key

untrust(keys: List[GpgKey])[source]¶

Delete known keys.

Parameters:: keys – keys to be deleted

verify(signature: str | Path, blob: str | Path, suppress_warnings: bool = False)[source]¶

Verify the signature on a blob.

Parameters:

signature – signature file (or clearsigned file)
blob – blob to be verified. If None, then signature is assumed to be a clearsigned file.
suppress_warnings – whether or not to suppress warnings from GnuPG

class spack.util.gpg.GpgKey(data: Dict[str, str])[source]¶

Bases: object

add(data: Dict[str, str])[source]¶: Add metadata to a key

expires_at: datetime | None¶

fpr: str¶

rev: List[GpgSignature]¶

sig: List[GpgSignature]¶

subkey: List[GpgKey]¶

uid: List[GpgUserId]¶

updated_at: datetime | None¶

class spack.util.gpg.GpgKeyAlgorithm(*values)[source]¶

Bases: Enum

Gpg Algormithms

ref. https://www.iana.org/assignments/openpgp/openpgp.xhtml#openpgp-public-key-algorithms

DH¶

DSA¶

EC¶

ECDSA¶

ED25519¶

ED448¶

EDDSA¶

ELGAMAL¶

ELGAMAL_EO¶

LIBGCRYPT¶

ML_DSA_65¶

ML_DSA_87¶

ML_KEM_1024¶

ML_KEM_786¶

RSA¶

RSA_EO¶

RSA_SO¶

SLH_DSA_SHAKE_128F¶

SLH_DSA_SHAKE_128S¶

SLH_DSA_SHAKE_256S¶

UNKNOWN¶

X25519¶

X448¶

class spack.util.gpg.GpgKeyCapability(*values)[source]¶

Bases: Enum

Gpg Capabilities

AUTHENTICATE¶

CERTIFY¶

DISABLED¶

ENCRYPT¶

SIGN¶

UNKNOWN¶

class spack.util.gpg.GpgKeyCompliance(*values)[source]¶

Bases: Enum

Gpg compliance codes

DE_VS¶

DE_VS_EXP¶

RFC4880BIS¶

UNKNOWN¶

VULN¶

class spack.util.gpg.GpgKeyTrust(*values)[source]¶

Bases: Enum

Gpg Trust normalized for Field 1 and Field 9

ERROR¶

EXPIRED¶

FULL¶

KNOWN¶

MARGINAL¶

NEVER¶

REVOKED¶

SPECIAL¶

ULTIMATE¶

UNDEFINED¶

UNKNOWN¶

property ownertrust: int¶: Return the ownertrust file integer corresponding to the GpgKeyTrust

class spack.util.gpg.GpgKeyType(*values)[source]¶

Bases: Flag

Gpg Key types

PUBLIC¶

PUBLIC_SUBKEY¶

REVOCATION¶

SECRET¶

SECRET_SUBKEY¶

SECRET_SUBKEY_ONLY¶

SUBKEY¶

class spack.util.gpg.GpgSigType(*values)[source]¶

Bases: Enum

Gpg Key signature types

REVOCATION¶

REVOCATION_SO¶

SIGNATURE¶

class spack.util.gpg.GpgSignature(data: Dict[str, str])[source]¶: Bases: object

class spack.util.gpg.GpgUserId(data: Dict[str, str])[source]¶: Bases: object

spack.util.gpg.PGP_CLEARSIG_HEADER¶: PGP cleartext signature header

spack.util.gpg.SOCKET_DIR¶: Socket directory required if a non default home directory is used

exception spack.util.gpg.SpackGPGError(message: str, long_message: str | None = None)[source]¶

Bases: SpackError

Class raised when GPG errors are detected.

spack.util.gpg.clear()[source]¶: Reset the global state to uninitialized.

spack.util.gpg.create(**kwargs)[source]¶: Create a new key pair.

spack.util.gpg.export_keys(location: str, keys: List[GpgKey], secret: bool = False)[source]¶

Export public keys to a location passed as argument.

Parameters:

location – where to export the keys
keys – keys to be exported
secret – whether to export secret keys or not

spack.util.gpg.extract_data_from_clearsig(data: str) → str[source]¶: Extract data from a gpg cleartext signed file

spack.util.gpg.extract_json_from_clearsig(data) → Dict[Any, Any][source]¶: Extract data from a gpg cleartext signed file as json

spack.util.gpg.extract_public_keys(keyfile: str)[source]¶

Extract the key ids from a file

Parameters:: keyfile – file with the public key

spack.util.gpg.glist(trusted: bool, signing: bool, fmt: str = 'default')[source]¶

List known keys.

Parameters:

trusted – if True list public keys
signing – if True list private keys
fmt – Key formatting string (default, colons, short, fpr)

spack.util.gpg.gnupghome_override(dir: str)[source]¶

Set the GNUPGHOME to a new location for this context.

Parameters:: dir – new value for GNUPGHOME

spack.util.gpg.init(gnupghome: str | None = None, force: bool = False)[source]¶: Initialize the global state for Gpg.

spack.util.gpg.is_clearsig(data: str) → bool[source]¶: Check if data is wrapped in a cleartext signature

spack.util.gpg.public_keys(*args) → List[GpgKey][source]¶: Return a list of fingerprints

spack.util.gpg.sign(key: str, file: str, output: str, clearsign: bool = False)[source]¶

Sign a file with a key.

Parameters:

key – key to be used to sign
file – file to be signed
output – output file (either the clearsigned file or the detached signature)
clearsign – if True wraps the document in an ASCII-armored signature, if False creates a detached signature

spack.util.gpg.signing_keys(*args) → List[GpgKey][source]¶: Return the keys that can be used to sign binaries.

spack.util.gpg.trust(keyfile: str, *, fprs: List[str] | None = None, yes_to_all: bool = False)[source]¶

Import a public key from a file and trust it.

Parameters:

keyfile – file with the public key
fprs – fingerprints of keys to trust.
yes_to_all – trust all keys in the file if True, otherwise ask for each key. Ignored if fprs is provided.

spack.util.gpg.untrust(signing: bool, *keys)[source]¶

Delete known keys.

Parameters:

signing – if True deletes the secret keys
*keys – keys to be deleted

spack.util.gpg.verify(signature: str, file: str | None = None, suppress_warnings: bool = False)[source]¶

Verify the signature on a file.

Parameters:

signature – signature of the file (or clearsigned file)
file – file to be verified. If None, then signature is assumed to be a clearsigned file.
suppress_warnings – whether or not to suppress warnings from GnuPG

spack.util.hash module¶

spack.util.hash.b32_hash(content)[source]¶: Return the b32 encoded sha1 hash of the input string as a string.

spack.util.hash.base32_prefix_bits(hash_string, bits)[source]¶: Return the first <bits> bits of a base32 string as an integer.

spack.util.lang module¶

spack.util.lang.ClassProperty¶: A type alias that represents either a classproperty descriptor or a constant value of the same type. This allows derived classes to override a computed class-level property with a constant value while retaining type compatibility.

class spack.util.lang.Const(value)[source]¶

Bases: object

Class level constant, raises when trying to set the attribute

value¶

class spack.util.lang.DeprecatedProperty(name: str)[source]¶

Bases: object

Data descriptor to error or warn when a deprecated property is accessed.

Derived classes must define a factory method to return an adaptor for the deprecated property, if the descriptor is not set to error.

error_lvl¶: 0 - Nothing 1 - Warning 2 - Error

factory(instance, owner)[source]¶

name¶

class spack.util.lang.GroupedExceptionForwarder(context: str, handler: GroupedExceptionHandler, base: type)[source]¶

Bases: object

A contextmanager to capture exceptions and forward them to a GroupedExceptionHandler.

class spack.util.lang.GroupedExceptionHandler[source]¶

Bases: object

A generic mechanism to coalesce multiple exceptions and preserve tracebacks.

exceptions: List[Tuple[str, Exception, List[str]]]¶

forward(context: str, base: type = BaseException) → GroupedExceptionForwarder[source]¶: Return a contextmanager which extracts tracebacks and prefixes a message.

grouped_message(with_tracebacks: bool = True) → str[source]¶: Print out an error message coalescing all the forwarded errors.

class spack.util.lang.HashableMap[source]¶

Bases: MutableMapping[K, V]

This is a hashable, comparable dictionary. Hash is performed on a tuple of the values in the dictionary.

dict: Dict[K, V]¶

class spack.util.lang.ObjectWrapper(wrapped_object)[source]¶

Bases: object

Base class that wraps an object. Derived classes can add new behavior while staying undercover.

This class is modeled after the stackoverflow answer: * http://stackoverflow.com/a/1445289/771663

class spack.util.lang.PriorityOrderedMapping[source]¶

Bases: Mapping[KT, VT]

Mapping that iterates over key according to an integer priority. If the priority is the same for two keys, insertion order is what matters.

The priority is set when the key/value pair is added. If not set, the highest current priority is used.

add(key: KT, *, value: VT, priority: int | None = None) → None[source]¶

Adds a key/value pair to the mapping, with a specific priority.

If the priority is None, then it is assumed to be the highest priority value currently in the container.

Raises:: ValueError – when the same priority is already in the mapping

priority_values(priority: int)[source]¶: Iterate over values of a given priority.

remove(key: KT) → VT[source]¶

Removes a key from the mapping.

Returns:: The value associated with the key being removed
Raises:: KeyError – if the key is not in the mapping

reversed_keys()[source]¶: Iterates over keys from the highest priority, to the lowest.

reversed_values()[source]¶: Iterates over values from the highest priority, to the lowest.

class spack.util.lang.Singleton(factory: Callable[[], object])[source]¶

Bases: object

Wrapper for lazily initialized singleton objects.

property instance¶

exception spack.util.lang.SingletonInstantiationError[source]¶

Bases: Exception

Error that indicates a singleton that cannot instantiate.

class spack.util.lang.TypedMutableSequence(iterable)[source]¶

Bases: MutableSequence

Base class that behaves like a list, just with a different type.

Client code can inherit from this base class:

class Foo(TypedMutableSequence):
    pass

and later perform checks based on types:

if isinstance(l, Foo):
    # do something

insert(index, item)[source]¶: S.insert(index, value) – insert value before index

exception spack.util.lang.UnhashableArguments[source]¶

Bases: TypeError

Raise when an @memoized function receives unhashable arg or kwarg values.

spack.util.lang.attr_setdefault(obj, name, value)[source]¶: Like dict.setdefault, but for objects.

class spack.util.lang.classproperty(callback: Callable[[Any], ClassPropertyType])[source]¶

Bases: Generic[ClassPropertyType]

Non-data descriptor to evaluate a class-level property. The function that performs the evaluation is injected at creation time and takes an owner (i.e., the class that originated the instance).

spack.util.lang.decorator_with_or_without_args(decorator)[source]¶

Allows a decorator to be used with or without arguments, e.g.:

# Calls the decorator function some args
@decorator(with, arguments, and=kwargs)

or:

# Calls the decorator function with zero arguments
@decorator

spack.util.lang.dedupe(sequence, key=None)[source]¶

Yields a stable de-duplication of an hashable sequence by key

Parameters:

sequence – hashable sequence to be de-duplicated
key – callable applied on values before uniqueness test; identity by default.

Returns:

stable de-duplication of the sequence

Examples

Dedupe a list of integers:

[x for x in dedupe([1, 2, 1, 3, 2])] == [1, 2, 3]

[x for x in spack.util.lang.dedupe([1,-2,1,3,2], key=abs)] == [1, -2, 3]

spack.util.lang.done¶: sentinel for testing that iterators are done in lazy_lexicographic_ordering

spack.util.lang.elide_list(line_list: List[str], max_num: int = 10) → List[str][source]¶

Takes a long list and limits it to a smaller number of elements, replacing intervening elements with "...". For example:

elide_list(["1", "2", "3", "4", "5", "6"], 4)

gives:

["1", "2", "3", "...", "6"]

spack.util.lang.ensure_last(lst, *elements)[source]¶: Performs a stable partition of lst, ensuring that elements occur at the end of lst in specified order. Mutates lst. Raises ValueError if any elements are not already in lst.

spack.util.lang.ensure_unwrapped(obj)[source]¶: Returns the real object behind a Singleton

spack.util.lang.fnmatch_translate_multiple(named_patterns: Dict[str, str]) → str[source]¶: Similar to fnmatch.translate, but takes an ordered dictionary where keys are pattern names, and values are filename patterns. The output is a regex that matches any of the patterns in order, and named capture groups are used to identify which pattern matched.

spack.util.lang.get_entry_points(*, group: str)[source]¶

Wrapper for importlib.metadata.entry_points

Parameters:: group – entry points to select
Returns:: EntryPoints for group or empty list if unsupported

spack.util.lang.index_by(objects, *funcs)[source]¶

Create a hierarchy of dictionaries by splitting the supplied set of objects on unique values of the supplied functions.

Values are used as keys. For example, suppose you have four objects with attributes that look like this:

a = Spec("boost %gcc target=skylake")
b = Spec("mrnet %intel target=zen2")
c = Spec("libelf %xlc target=skylake")
d = Spec("libdwarf %intel target=zen2")

list_of_specs = [a,b,c,d]
index1 = index_by(list_of_specs, lambda s: str(s.target),
                  lambda s: s.compiler)
index2 = index_by(list_of_specs, lambda s: s.compiler)

index1 now has two levels of dicts, with lists at the leaves, like this:

{ 'zen2'    : { 'gcc' : [a], 'xlc' : [c] },
  'skylake' : { 'intel' : [b, d] }
}

And index2 is a single level dictionary of lists that looks like this:

{ 'gcc'    : [a],
  'intel'  : [b,d],
  'xlc'    : [c]
}

If any elements in funcs is a string, it is treated as the name of an attribute, and acts like getattr(object, name). So shorthand for the above two indexes would be:

index1 = index_by(list_of_specs, 'arch', 'compiler')
index2 = index_by(list_of_specs, 'compiler')

You can also index by tuples by passing tuples:

index1 = index_by(list_of_specs, ('target', 'compiler'))

Keys in the resulting dict will look like ('gcc', 'skylake').

spack.util.lang.key_ordering(cls)[source]¶

Decorates a class with extra methods that implement rich comparison operations and __hash__. The decorator assumes that the class implements a function called _cmp_key(). The rich comparison operations will compare objects using this key, and the __hash__ function will return the hash of this key.

If a class already has __eq__, __ne__, __lt__, __le__, __gt__, or __ge__ defined, this decorator will overwrite them.

Raises:: TypeError – If the class does not have a _cmp_key method

spack.util.lang.lazy_eq(lseq, rseq)[source]¶

Equality comparison for two lazily generated sequences.

See lazy_lexicographic_ordering.

spack.util.lang.lazy_lexicographic_ordering(cls, set_hash=True)[source]¶

Decorates a class with extra methods that implement rich comparison.

This is a lazy version of the tuple comparison used frequently to implement comparison in Python. Given some objects with fields, you might use tuple keys to implement comparison, e.g.:

class Widget:
    def _cmp_key(self):
        return (self.a, self.b, (self.c, self.d), self.e)

    def __eq__(self, other):
        return self._cmp_key() == other._cmp_key()

    def __lt__(self):
        return self._cmp_key() < other._cmp_key()

    # etc.

Python would compare Widgets lexicographically based on their tuples. The issue there for simple comparators is that we have to build the tuples and we have to generate all the values in them up front. When implementing comparisons for large data structures, this can be costly.

Lazy lexicographic comparison maps the tuple comparison shown above to generator functions. Instead of comparing based on pre-constructed tuple keys, users of this decorator can compare using elements from a generator. So, you’d write:

@lazy_lexicographic_ordering
class Widget:
    def _cmp_iter(self):
        yield a
        yield b
        def cd_fun():
            yield c
            yield d
        yield cd_fun
        yield e

    # operators are added by decorator

There are no tuples preconstructed, and the generator does not have to complete. Instead of tuples, we simply make functions that lazily yield what would’ve been in the tuple. The @lazy_lexicographic_ordering decorator handles the details of implementing comparison operators, and the Widget implementor only has to worry about writing _cmp_iter, and making sure the elements in it are also comparable.

In some cases, you may have a fast way to determine whether two objects are equal, e.g. the is function or an already-computed cryptographic hash. For this, you can implement your own _cmp_fast_eq function:

@lazy_lexicographic_ordering
class Widget:
    def _cmp_iter(self):
        yield a
        yield b

        def cd_fun():
            yield c
            yield d

        yield cd_fun
        yield e

    def _cmp_fast_eq(self, other):
        return self is other or None

_cmp_fast_eq should return:

True if self is equal to other,
False if self is not equal to other, and
None if it’s not known whether they are equal, and the full comparison should be done.

lazy_lexicographic_ordering uses _cmp_fast_eq to short-circuit the comparison if the answer can be determined quickly. If you do not implement it, it defaults to self is other or None.

Some things to note:

If a class already has __eq__, __ne__, __lt__, __le__, __gt__, __ge__, or __hash__ defined, this decorator will overwrite them.
If set_hash is False, this will not overwrite __hash__.
This class uses Python 2 None-comparison semantics. If you yield None and it is compared to a non-None type, None will always be less than the other object.

Raises:: TypeError – If the class does not have a _cmp_iter method

spack.util.lang.lazy_lt(lseq, rseq)[source]¶

Less-than comparison for two lazily generated sequences.

See lazy_lexicographic_ordering.

spack.util.lang.list_modules(directory, **kwargs)[source]¶: Lists all of the modules, excluding __init__.py, in a particular directory. Listed packages have no particular order.

spack.util.lang.load_module_from_file(module_name, module_path)[source]¶

Loads a python module from the path of the corresponding file.

If the module is already in sys.modules it will be returned as is and not reloaded.

Parameters:

module_name (str) – namespace where the python module will be loaded, e.g. foo.bar
module_path (str) – path of the python file containing the module

Returns:

A valid module object

Raises:

ImportError – when the module can’t be loaded
FileNotFoundError – when module_path doesn’t exist

spack.util.lang.match_predicate(*args)[source]¶

Utility function for making string matching predicates.

Each arg can be a: * regex * list or tuple of regexes * predicate that takes a string.

This returns a predicate that is true if: * any arg regex matches * any regex in a list or tuple of regexes matches. * any predicate in args matches.

spack.util.lang.memoized(func)[source]¶

Decorator that caches the results of a function, storing them in an attribute of that function.

Example:

@memoized
def expensive_computation(x):
    # Some expensive computation
    return result

spack.util.lang.nullcontext(*args, **kwargs)[source]¶: Empty context manager. TODO: replace with contextlib.nullcontext() if we ever require python 3.7.

spack.util.lang.pretty_date(time: datetime | int, now: datetime | None = None) → str[source]¶

Convert a datetime or timestamp to a pretty, relative date.

Parameters:

time – date to print prettily
now – the date the pretty date is relative to (default is datetime.now())

Returns:

pretty string like “an hour ago”, “Yesterday”, “3 months ago”, “just now”, etc.

Adapted from https://stackoverflow.com/questions/1551382.

spack.util.lang.pretty_duration(seconds: float) → str[source]¶: Format a duration in seconds as a compact human-readable string (e.g. “1h02m”, “3m05s”, “45s”).

spack.util.lang.pretty_seconds(seconds)[source]¶

Seconds to string with appropriate units

Parameters:: seconds (float) – Number of seconds
Returns:: Time string with units
Return type:: str

spack.util.lang.pretty_seconds_formatter(seconds)[source]¶

spack.util.lang.pretty_string_to_date(date_str: str, now: datetime | None = None) → datetime[source]¶

Parses a string representing a date and returns a datetime object.

Parameters:: date_str – string representing a date. This string might be in different format (like YYYY, YYYY-MM, YYYY-MM-DD, YYYY-MM-DD HH:MM, YYYY-MM-DD HH:MM:SS) or be a pretty date (like yesterday or two months ago)

Returns: datetime object corresponding to date_str

spack.util.lang.stable_partition(input_iterable: Iterable[T], predicate_fn: Callable[[T], bool]) → Tuple[List[T], List[T]][source]¶

Partition the input iterable according to a custom predicate.

Parameters:

input_iterable – input iterable to be partitioned.
predicate_fn – predicate function accepting an iterable item as argument.

Returns:

Tuple of the list of elements evaluating to True, and list of elements evaluating to False.

spack.util.lang.tuplify(seq)[source]¶: Helper for lazy_lexicographic_ordering().

spack.util.lang.uniq(sequence)[source]¶

Remove strings of duplicate elements from a list.

This works like the command-line uniq tool. It filters strings of duplicate elements in a list. Adjacent matching elements are merged into the first occurrence.

For example:

uniq([1, 1, 1, 1, 2, 2, 2, 3, 3]) == [1, 2, 3]
uniq([1, 1, 1, 1, 2, 2, 2, 1, 1]) == [1, 2, 1]

spack.util.ld_so_conf module¶

spack.util.ld_so_conf.get_conf_file_from_dynamic_linker(dynamic_linker_name)[source]¶

spack.util.ld_so_conf.host_dynamic_linker_search_paths()[source]¶: Retrieve the current host runtime search paths for shared libraries; for GNU and musl Linux we try to retrieve the dynamic linker from the current Python interpreter and then find the corresponding config file (e.g. ld.so.conf or ld-musl-<arch>.path). Similar can be done for BSD and others, but this is not implemented yet. The default paths are always returned. We don’t check if the listed directories exist.

spack.util.ld_so_conf.parse_ld_so_conf(conf_file='/etc/ld.so.conf')[source]¶

Parse glibc style ld.so.conf file, which specifies default search paths for the dynamic linker. This can in principle also be used for musl libc.

Parameters:: conf_file (str or bytes) – Path to config file
Returns:: List of absolute search paths
Return type:: list

spack.util.libc module¶

spack.util.libc.GLIBC_PATTERN¶: Pattern to distinguish glibc from other libc implementations

spack.util.libc.default_search_paths_from_dynamic_linker(dynamic_linker: str) → List[str][source]¶: If the dynamic linker is glibc at a certain version, we can query the hard-coded library search paths

spack.util.libc.libc_from_current_python_process() → Spec | None[source]¶

spack.util.libc.libc_from_dynamic_linker(dynamic_linker: str) → Spec | None[source]¶: Get the libc spec from the dynamic linker path.

spack.util.libc.parse_dynamic_linker(output: str)[source]¶: Parse -dynamic-linker /path/to/ld.so from compiler output

spack.util.link_tree module¶

LinkTree class for setting up trees of symbolic links.

class spack.util.link_tree.LinkTree(source_root)[source]¶

Bases: object

Class to create trees of symbolic links from a source directory.

LinkTree objects are constructed with a source root. Their methods allow you to create and delete trees of symbolic links back to the source tree in specific destination directories. Trees comprise symlinks only to files; directories are never symlinked to, to prevent the source directory from ever being modified.

find_conflict(dest_root, ignore=None, ignore_file_conflicts=False)[source]¶: Returns the first file in dest that conflicts with src

find_dir_conflicts(dest_root, ignore)[source]¶

get_file_map(dest_root, ignore)[source]¶

merge(dest_root, ignore_conflicts: bool = False, ignore: Callable[[str], bool] | None = None, link: Callable = fs.symlink, relative: bool = False)[source]¶

Link all files in src into dest, creating directories if necessary.

Parameters:

ignore_conflicts – if True, do not break when the target exists; return a list of files that could not be linked
ignore – callable that returns True if a file is to be ignored in the merge (by default ignore nothing)
link – function to create links with (defaults to spack.util.filesystem.symlink)
relative – create all symlinks relative to the target (default False)

merge_directories(dest_root, ignore)[source]¶

unmerge(dest_root, ignore=None, remove_file=remove_link)[source]¶

Unlink all files in dest that exist in src.

Unlinks directories in dest if they are empty.

unmerge_directories(dest_root, ignore)[source]¶

spack.util.lock module¶

exception spack.util.lock.CantCreateLockError(path: str)[source]¶

Bases: LockPermissionError

Attempt to create a lock in an unwritable location.

class spack.util.lock.DummyBackend[source]¶

Bases: object

No-op lock backend: all operations succeed without acquiring any real locks.

cleanup(path: str) → None[source]¶

poll(op: int) → bool[source]¶

prepare(op: int) → None[source]¶

release() → None[source]¶

class spack.util.lock.Lock(path: str, *, start: int = 0, length: int = 0, default_timeout: float | None = None, debug: bool = False, desc: str = '', enable: bool = True)[source]¶

Bases: object

This is an implementation of a filesystem lock using Python’s lockf.

In Python, lockf actually calls fcntl, so this should work with any filesystem implementation that supports locking through the fcntl calls. This includes distributed filesystems like Lustre (when flock is enabled) and recent NFS versions.

Note that this is for managing contention over resources between processes and not for managing contention between threads in a process: the functions of this object are not thread-safe. A process also must not maintain multiple locks on the same file (or, more specifically, on overlapping byte ranges in the same file).

acquire_read(timeout: float | None = None) → bool[source]¶

Acquires a recursive, shared lock for reading.

Read and write locks can be acquired and released in arbitrary order, but the POSIX lock is held until all local read and write locks are released.

Returns True if it is the first acquire and actually acquires the POSIX lock, False if it is a nested transaction.

acquire_write(timeout: float | None = None) → bool[source]¶

Acquires a recursive, exclusive lock for writing.

Read and write locks can be acquired and released in arbitrary order, but the POSIX lock is held until all local read and write locks are released.

Returns True if it is the first acquire and actually acquires the POSIX lock, False if it is a nested transaction.

cleanup() → None[source]¶

downgrade_write_to_read(timeout: float | None = None) → None[source]¶

Downgrade from an exclusive write lock to a shared read.

Raises:: LockDowngradeError – if this is an attempt at a nested transaction

is_write_locked() → bool[source]¶: Returns True if the path is write locked, otherwise, False

release_read(release_fn: Callable[[], bool | None] | None = None) → bool[source]¶

Releases a read lock.

Parameters:: release_fn – function to call before the last recursive lock (read or write) is released.

If the last recursive lock will be released, then this will call release_fn and return its result (if provided), or return True (if release_fn was not provided).

Otherwise, we are still nested inside some other lock, so do not call the release_fn and, return False.

Does limited correctness checking: if a read lock is released when none are held, this will raise an assertion error.

release_write(release_fn: Callable[[], bool | None] | None = None) → bool[source]¶

Releases a write lock.

Parameters:: release_fn – function to call before the last recursive write is released.

If the last recursive write lock will be released, then this will call release_fn and return its result (if provided), or return True (if release_fn was not provided). Otherwise, we are still nested inside some other write lock, so do not call the release_fn, and return False.

Does limited correctness checking: if a read lock is released when none are held, this will raise an assertion error.

try_acquire_read() → bool[source]¶

Non-blocking attempt to acquire a shared read lock.

Returns True if the lock was acquired, False if it would block.

try_acquire_write() → bool[source]¶

Non-blocking attempt to acquire an exclusive write lock.

Returns True if the lock was acquired, False if it would block.

upgrade_read_to_write(timeout: float | None = None) → None[source]¶

Attempts to upgrade from a shared read lock to an exclusive write.

Raises:: LockUpgradeError – if this is an attempt at a nested transaction

exception spack.util.lock.LockDowngradeError(path: str)[source]¶

Bases: LockError

Raised when unable to downgrade from a write to a read lock.

exception spack.util.lock.LockError[source]¶

Bases: Exception

Raised for any errors related to locks.

exception spack.util.lock.LockPermissionError[source]¶

Bases: LockError

Raised when there are permission issues with a lock.

exception spack.util.lock.LockROFileError(path: str)[source]¶

Bases: LockPermissionError

Tried to take an exclusive lock on a read-only file.

exception spack.util.lock.LockTimeoutError(lock_type: int, path: str, time: float, attempts: int)[source]¶

Bases: LockError

Raised when an attempt to acquire a lock times out.

Bases: object

Simple nested transaction context manager that uses a file lock.

Parameters:

lock – underlying lock for this transaction to be acquired on enter and released on exit
acquire – function to be called after lock is acquired
release – function to be called before release, with (exc_type, exc_value, traceback)
timeout – number of seconds to set for the timeout when acquiring the lock (default no timeout)

exception spack.util.lock.LockUpgradeError(path: str)[source]¶

Bases: LockError

Raised when unable to upgrade from a read to a write lock.

class spack.util.lock.PosixBackend(path: str, start: int, length: int, debug: bool = False)[source]¶

Bases: object

fcntl-based lock backend for POSIX systems.

cleanup(path: str) → None[source]¶: Remove the lock file.

poll(op: int) → bool[source]¶: Attempt to acquire the lock in a non-blocking manner. Return whether the locking attempt succeeds

prepare(op: int) → None[source]¶: Ensure the lock file is open; raise if a write lock is requested on a read-only file.

release() → None[source]¶

Releases a lock using POSIX locks (fcntl.lockf)

Releases the lock regardless of mode. Note that read locks may be masquerading as write locks, but this removes either.

Bases: LockTransaction

LockTransaction context manager that does a read and releases it.

Bases: ReadTransaction

Non-blocking ReadTransaction: yields True if the lock was acquired, and False if acquiring it would block, in which case the body must skip its work:

with TryReadTransaction(lock, acquire=...) as acquired:
    if not acquired:
        return
    ...

Bases: WriteTransaction

Non-blocking WriteTransaction: yields True if the lock was acquired, and False if acquiring it would block, in which case the body must skip its work:

with TryWriteTransaction(lock, acquire=..., release=...) as acquired:
    if not acquired:
        return
    ...

Bases: LockTransaction

LockTransaction context manager that does a write and releases it.

spack.util.lock.check_lock_safety(path: str) → None[source]¶

Do some extra checks to ensure disabling locks is safe.

This will raise an error if path can is group- or world-writable AND the current user can write to the directory (i.e., if this user AND others could write to the path).

This is intended to run on the Spack prefix, but can be run on any path for testing.

spack.util.log_parse module¶

spack.util.log_parse.make_log_context(log_events: List[LogEvent]) → str[source]¶

Get error context from a log file.

Parameters:: log_events – list of events created by ctest_log_parser.parse()
Returns:: context from the build log with errors highlighted
Return type:: str

Parses the log file for lines containing errors, and prints them out with context. Errors are highlighted in red and warnings in yellow. Events are sorted by line number.

spack.util.log_parse.parse_log_events(stream: str | TextIO, context: int = 6, profile: bool = False, tail: int = 0) → Tuple[List[BuildError], List[BuildWarning], LogEvent | None][source]¶

Extract interesting events from a log file.

Parameters:

stream – build log name or file object
context – lines of context to extract around each log event
profile – print out profile information for parsing
tail – if > 0, also return the last tail lines

Returns:

two lists containing BuildError and BuildWarning objects, plus an optional LogEvent for the tail (None when tail=0).

spack.util.module_cmd module¶

This module contains routines related to the module command for accessing and parsing environment modules.

exception spack.util.module_cmd.ModuleLoadError(module)[source]¶

Bases: SpackError

Raised when a module cannot be loaded.

spack.util.module_cmd.get_path_args_from_module_line(line)[source]¶

spack.util.module_cmd.get_path_from_module_contents(text, module_name)[source]¶

spack.util.module_cmd.load_module(mod)[source]¶

Takes a module name and removes modules until it is possible to load that module. It then loads the provided module. Depends on the modulecmd implementation of modules used in cray and lmod.

Raises:: ModuleLoadError – if the module could not be loaded

spack.util.module_cmd.module(*args: str, module_template: str | None = None, module_src_cmd: str | None = None, environb: MutableMapping[bytes, bytes] | None = None)[source]¶

Run the module shell function in a /bin/bash subprocess, and either collect its changes to environment variables and apply them in the current process (for module load, module swap, etc.), or return its output as a string (for module show, etc.).

This requires /bin/bash to be available on the system and awk to be in PATH.

Parameters:

args – Command line arguments for the module command.
environb – (Binary) environment variables dictionary. If not provided, the current process’s environment is modified.

spack.util.module_cmd.path_from_modules(modules)[source]¶

Inspect a list of Tcl modules for entries that indicate the absolute path at which the library supported by said module can be found.

Parameters:: modules (list) – module files to be loaded to get an external package
Returns:: Guess of the prefix path where the package

spack.util.naming module¶

class spack.util.naming.NamespaceTrie(separator='.')[source]¶

Bases: object

class Element(value)[source]¶: Bases: object

has_value(namespace)[source]¶: True if there is a value set for the given namespace.

is_leaf(namespace)[source]¶: True if this namespace has no children in the trie.

is_prefix(namespace)[source]¶: True if the namespace has a value, or if it’s the prefix of one that does.

spack.util.naming.pkg_name_to_class_name(pkg_name: str)[source]¶

Convert a Spack package name to a class name, based on PEP-8:

Module and package names use lowercase_with_underscores.
Class names use the CapWords convention.

Not all package names are valid Python identifiers:

They can contain -, but cannot start with -.
They can start with numbers, e.g. 3proxy.

This function converts from the package name to the class convention by removing _ and -, and converting surrounding lowercase text to CapWords. If package name starts with a number, the class name returned will be prepended with _ to make a valid Python identifier.

spack.util.naming.possible_spack_module_names(python_mod_name: str) → List[str][source]¶: Given a Python module name, return a list of all possible spack module names that could correspond to it.

spack.util.naming.simplify_name(name: str) → str[source]¶

Simplify package name to only lowercase, digits, and dashes.

Simplifies a name which may include uppercase letters, periods, underscores, and pluses. In general, we want our package names to only contain lowercase letters, digits, and dashes.

Parameters:: name (str) – The original name of the package
Returns:: The new name of the package
Return type:: str

spack.util.naming.valid_module_name(mod_name: str, package_api: Tuple[int, int]) → bool[source]¶: Return whether mod_name is valid for use in Spack.

spack.util.package_hash module¶

exception spack.util.package_hash.PackageHashError(message: str, long_message: str | None = None)[source]¶

Bases: SpackError

Raised for all errors encountered during package hashing.

class spack.util.package_hash.RemoveDirectives(spec)[source]¶

Bases: NodeTransformer

Remove Spack directives from a package AST.

This removes Spack directives (e.g., depends_on, conflicts, etc.) and metadata attributes (e.g., tags, homepage, url) in a top-level class definition within a package.py, but it does not modify nested classes or functions.

If removing directives causes a for, with, or while statement to have an empty body, we remove the entire statement. Similarly, If removing directives causes an if statement to have an empty body or else block, we’ll remove the block (or replace the body with pass if there is an else block but no body).

metadata_attrs¶: List of attributes to be excluded from a package’s hash.

visit_Assign(node)[source]¶

visit_ClassDef(node)[source]¶

visit_Expr(node)[source]¶

visit_For(node)[source]¶

visit_FunctionDef(node)[source]¶

visit_If(node)[source]¶

visit_While(node)[source]¶

visit_With(node)[source]¶

class spack.util.package_hash.RemoveDocstrings[source]¶

Bases: NodeTransformer

Transformer that removes docstrings from a Python AST.

This removes all strings that aren’t on the RHS of an assignment statement from the body of functions, classes, and modules – even if they’re not directly after the declaration.

remove_docstring(node)[source]¶

visit_ClassDef(node)[source]¶

visit_FunctionDef(node)[source]¶

visit_Module(node)[source]¶

class spack.util.package_hash.ResolveMultiMethods(methods)[source]¶

Bases: NodeTransformer

Remove multi-methods when we know statically that they won’t be used.

Say we have multi-methods like this:

class SomePackage:
    def foo(self): print("implementation 1")

    @when("@1.0")
    def foo(self): print("implementation 2")

    @when("@2.0")
    @when(sys.platform == "darwin")
    def foo(self): print("implementation 3")

    @when("@3.0")
    def foo(self): print("implementation 4")

The multimethod that will be chosen at runtime depends on the package spec and on whether we’re on the darwin platform at build time (the darwin condition for implementation 3 is dynamic). We know the package spec statically; we don’t know statically what the runtime environment will be. We need to include things that can possibly affect package behavior in the package hash, and we want to exclude things when we know that they will not affect package behavior.

If we’re at version 4.0, we know that implementation 1 will win, because some @when for 2, 3, and 4 will be False. We should only include implementation 1.

If we’re at version 1.0, we know that implementation 2 will win, because it overrides implementation 1. We should only include implementation 2.

If we’re at version 3.0, we know that implementation 4 will win, because it overrides implementation 1 (the default), and some @when on all others will be False.

If we’re at version 2.0, it’s a bit more complicated. We know we can remove implementations 2 and 4, because their @when’s will never be satisfied. But, the choice between implementations 1 and 3 will happen at runtime (this is a bad example because the spec itself has platform information, and we should prefer to use that, but we allow arbitrary boolean expressions in @when’s, so this example suffices). For this case, we end up needing to include both implementation 1 and 3 in the package hash, because either could be chosen.

resolve(impl_conditions)[source]¶: Given list of nodes and conditions, figure out which node will be chosen.

visit_FunctionDef(node: FunctionDef) → FunctionDef | None[source]¶

class spack.util.package_hash.TagMultiMethods(spec: Spec)[source]¶

Bases: NodeVisitor

Tag @when-decorated methods in a package AST.

methods: Dict[str, List[Tuple[FunctionDef, List[bool | None]]]]¶

visit_FunctionDef(node: FunctionDef) → AST[source]¶

spack.util.package_hash.canonical_source(spec: Spec, filter_multimethods: bool = True, source: bytes | None = None) → str[source]¶

Get canonical source for a spec’s package.py by unparsing its AST.

Parameters:

filter_multimethods – By default, filter multimethods out of the AST if they are known statically to be unused. Supply False to disable.
source – Optionally provide a string to read python code from.

spack.util.package_hash.package_ast(spec: Spec, filter_multimethods: bool = True, source: bytes | None = None) → AST[source]¶

Get the AST for the package.py file corresponding to spec.

Parameters:

filter_multimethods – By default, filter multimethods out of the AST if they are known statically to be unused. Supply False to disable.
source – Optionally provide a string to read python code from.

spack.util.package_hash.package_hash(spec: Spec, source: bytes | None = None) → str[source]¶

Get a hash of a package’s canonical source code.

This function is used to determine whether a spec needs a rebuild when a package’s source code changes.

Parameters:: source – Optionally provide a string to read python code from.

spack.util.package_hash.unused_string(node: AST) → bool[source]¶: Criteria for unassigned body strings.

spack.util.parallel module¶

spack.util.parallel.ENABLE_PARALLELISM¶: Used in tests to disable parallelism, as tests themselves are parallelized

class spack.util.parallel.ErrorFromWorker(exc_cls, exc, tb)[source]¶

Bases: object

Wrapper class to report an error from a worker process

property stacktrace¶

class spack.util.parallel.SequentialExecutor[source]¶

Bases: Executor

Executor that runs tasks sequentially in the current thread.

submit(fn, *args, **kwargs)[source]¶: Submit a function to be executed.

class spack.util.parallel.Task(func)[source]¶

Bases: object

Wrapped task that trap every Exception and return it as an ErrorFromWorker object.

We are using a wrapper class instead of a decorator since the class is pickleable, while a decorator with an inner closure is not.

spack.util.parallel.imap_unordered(f, list_of_args, *, processes: int, maxtaskperchild: int | None = None, debug=False, serialize_env: bool = False)[source]¶

Wrapper around multiprocessing.Pool.imap_unordered.

Parameters:

f – function to apply
list_of_args – list of tuples of args for the task
processes – maximum number of processes allowed
debug – if False, raise an exception containing just the error messages from workers, if True an exception with complete stacktraces
maxtaskperchild – number of tasks to be executed by a child before being killed and substituted

Raises:

RuntimeError – if any error occurred in the worker processes

spack.util.parallel.make_concurrent_executor(jobs: int | None = None, *, serialize_env: bool = False) → Executor[source]¶

Create a concurrent executor.

If serialize_env is False (default), the active Spack environment is not transmitted to the worker processes, which avoids the cost of pickling potentially large environment state.

spack.util.path module¶

Utilities for managing paths in Spack.

TODO: this is really part of spack.config. Consolidate it.

spack.util.pattern module¶

class spack.util.pattern.Args(*flags, **kwargs)[source]¶

Bases: Bunch

Subclass of Bunch to write argparse args more naturally.

class spack.util.pattern.Bunch(**kwargs)[source]¶

Bases: object

Carries a bunch of named attributes (from Alex Martelli bunch)

spack.util.prefix module¶

This file contains utilities for managing the installation prefix of a package.

class spack.util.prefix.Prefix[source]¶

Bases: str

This class represents an installation prefix, but provides useful attributes for referring to directories inside the prefix.

Attributes of this object are created on the fly when you request them, so any of the following are valid:

>>> prefix = Prefix("/usr")
>>> prefix.bin
/usr/bin
>>> prefix.lib64
/usr/lib64
>>> prefix.share.man
/usr/share/man
>>> prefix.foo.bar.baz
/usr/foo/bar/baz
>>> prefix.join("dashed-directory").bin64
/usr/dashed-directory/bin64

Prefix objects behave identically to strings. In fact, they subclass str, so operators like + are legal:

print("foobar " + prefix)

This prints foobar /usr. All of this is meant to make custom installs easy.

join(string: str) → Prefix[source]¶

Concatenate a string to a prefix.

Useful for strings that are not valid variable names. This includes strings containing characters like - and ..

Parameters:: string – the string to append to the prefix
Returns:: the newly created installation prefix

spack.util.remote_file_cache module¶

spack.util.remote_file_cache.fetch_remote_text_file(url: str, dest_dir: str) → str[source]¶

Retrieve the text file from the url into the destination directory.

Parameters:

url – URL for the remote text file
dest_dir – destination directory in which to stage the file locally

Returns:

Path to the fetched file

Raises:

ValueError – if there are missing required arguments

spack.util.remote_file_cache.local_path(path: str, sha256: str, dest: str | None = None) → str[source]¶

Determine the actual path and, if remote, stage its contents locally.

Parameters:

path – the resolved configuration path
sha256 – the expected sha256 if the file is remote
dest – destination path

Returns: normalized local path

Raises:: ValueError – missing or mismatched arguments, unsupported URL scheme

spack.util.remote_file_cache.raw_github_gitlab_url(url: str) → str[source]¶

Transform a github URL to the raw form to avoid undesirable html.

Parameters:: url – url to be converted to raw form
Returns:: Raw github/gitlab url or the original url

spack.util.s3 module¶

class spack.util.s3.UrllibS3Handler[source]¶

Bases: BaseHandler

s3_open(req)[source]¶

class spack.util.s3.WrapStream(raw)[source]¶

Bases: BufferedReader

detach()[source]¶

Disconnect this buffer from its underlying raw stream and return it.

After the raw stream has been detached, the buffer is in an unusable state.

read(*args, **kwargs)[source]¶

Read and return up to n bytes.

If the size argument is omitted, None, or negative, read and return all data until EOF.

Return an empty bytes object on EOF.

Return None if the underlying raw stream was open in non-blocking mode and no data is available at the moment.

spack.util.s3.get_mirror_s3_connection_info(mirror, method)[source]¶: Create s3 config for session/client from a Mirror instance (or just set defaults when no mirror is given.)

spack.util.s3.get_s3_session(url, method='fetch')[source]¶

spack.util.s3.s3_client_cache: Dict[Tuple[str, str], Any]¶: Map (mirror name, method) tuples to s3 client instances.

spack.util.socket module¶

Optimized Spack implementations of methods from socket module.

spack.util.spack_json module¶

Simple wrapper around JSON to guarantee consistent use of load/dump.

exception spack.util.spack_json.SpackJSONError(msg: str, json_error: BaseException)[source]¶

Bases: SpackError

Raised when there are issues with JSON parsing.

spack.util.spack_json.dump(data: Any, stream: IO[str], pretty: bool = False) → None[source]¶: Wrapper around json.dump with different default arguments

spack.util.spack_json.dumps(data: Any, pretty: bool = False) → str[source]¶: Wrapper around json.dumps with different default arguments

spack.util.spack_json.load(stream: Any) → Dict[source]¶: Spack JSON needs to be ordered to support specs.

spack.util.spack_yaml module¶

Enhanced YAML parsing for Spack.

load() preserves YAML Marks on returned objects – this allows us to access file and line information later.
Our load methods use ``OrderedDict class instead of YAML’s default unordered dict.

exception spack.util.spack_yaml.SpackYAMLError(msg, yaml_error, filename=None)[source]¶

Bases: SpackError

Raised when there are issues with YAML parsing.

spack.util.spack_yaml.dump(data, stream=None, default_flow_style=False)[source]¶

spack.util.spack_yaml.load(*args, **kwargs)[source]¶

spack.util.string module¶

String manipulation functions that do not have other dependencies than Python standard library

spack.util.string.comma_and(sequence: List[str]) → str[source]¶: Return a string with all the elements of the input joined by comma, but the last one (which is joined by "and").

spack.util.string.comma_list(sequence: Sequence[str], article: str = '') → str[source]¶

spack.util.string.comma_or(sequence: Sequence[str]) → str[source]¶: Return a string with all the elements of the input joined by comma, but the last one (which is joined by "or").

spack.util.string.ordinal(number: int) → str[source]¶

Return the ordinal representation (1st, 2nd, 3rd, etc.) for the provided number.

Parameters:: number – int to convert to ordinal number

Returns: number’s corresponding ordinal

spack.util.string.plural(n: int, singular: str, plural: str | None = None, show_n: bool = True) → str[source]¶

Pluralize <singular> word by adding an s if n != 1.

Parameters:

n – number of things there are
singular – singular form of word
plural – optional plural form, for when it’s not just singular + ‘s’
show_n – whether to include n in the result string (default True)

Returns:

“1 thing” if n == 1 or “n things” if n != 1

spack.util.string.quote(sequence: List[str], q: str = "'") → List[str][source]¶: Quotes each item in the input list with the quote character passed as second argument.

spack.util.timer module¶

Debug signal handler: prints a stack trace and enters interpreter.

register_interrupt_handler() enables a ctrl-C handler that prints a stack trace and drops the user into an interpreter.

class spack.util.timer.BaseTimer[source]¶

Bases: object

duration(name=None)[source]¶

measure(name)[source]¶

property phases¶

start(name=None)[source]¶

stop(name=None)[source]¶

write_json(out=sys.stdout)[source]¶

write_tty(out=sys.stdout)[source]¶

spack.util.timer.NULL_TIMER¶: instance of a do-nothing timer

class spack.util.timer.NullTimer[source]¶

Bases: BaseTimer

Timer interface that does nothing, useful in for “tell don’t ask” style code when timers are optional.

class spack.util.timer.TimeTracker(total, start, count, path)¶

Bases: tuple

count¶: Alias for field number 2

path¶: Alias for field number 3

start¶: Alias for field number 1

total¶: Alias for field number 0

class spack.util.timer.Timer(now: Callable[[], float] = time.time)[source]¶

Bases: BaseTimer

Simple interval timer

duration(name=global_timer_name)[source]¶

Get the time in seconds of a named timer, or the total time if no name is passed. The duration is always 0 for timers that have not been started, no error is raised.

Parameters:: name (str) – (Optional) name of the timer
Returns:: duration of timer.
Return type:: float

measure(name)[source]¶

Context manager that allows you to time a block of code.

Parameters:: name (str) – Name of the timer

property phases¶: Get all named timers (excluding the global/total timer)

start(name=global_timer_name)[source]¶

Start or restart a named timer, or the global timer when no name is given.

Parameters:: name (str) – Optional name of the timer. When no name is passed, the global timer is started.

stop(name=global_timer_name)[source]¶

Stop a named timer, or all timers when no name is given. Stopping a timer that has not started has no effect.

Parameters:: name (str) – Optional name of the timer. When no name is passed, all timers are stopped.

write_json(out=sys.stdout, extra_attributes={})[source]¶: Write a json object with times to file

write_tty(out=sys.stdout)[source]¶: Write a human-readable summary of timings (depth is 1)

class spack.util.timer.TimerEvent(time, running, label)¶

Bases: tuple

label¶: Alias for field number 2

running¶: Alias for field number 1

time¶: Alias for field number 0

spack.util.timer.global_timer_name¶: name for the global timer (used in start(), stop(), duration() without arguments)

spack.util.typing module¶

Extra support for type checking in Spack.

Protocols here that have runtime overhead should be set to object when TYPE_CHECKING is not enabled, as they can incur unreasonable runtime overheads.

In particular, Protocols intended for use on objects that have many isinstance() calls can be very expensive.

spack.util.url module¶

Utility functions for parsing, formatting, and manipulating URLs.

spack.util.url.allowed_archive(path_or_url: str) → bool[source]¶: Returns true if the input is a valid archive, False otherwise.

spack.util.url.check_and_remove_ext(path: str, *, extension: str) → str[source]¶: Returns the input path with the extension removed, if the extension is present in path. Otherwise, returns the input unchanged.

spack.util.url.compression_ext_from_compressed_archive(extension: str) → str | None[source]¶: Returns compression extension for a compressed archive

spack.util.url.default_download_filename(url: str) → str[source]¶: This method computes a default file name for a given URL. Note that it makes no request, so this is not the same as the option curl -O, which uses the remote file name from the response header.

spack.util.url.determine_url_file_extension(path: str) → str[source]¶

This returns the type of archive a URL refers to. This is sometimes confusing because of URLs like:

https://github.com/petdance/ack/tarball/1.93_02

Where the URL doesn’t actually contain the filename. We need to know what type it is so that we can appropriately name files in mirrors.

spack.util.url.expand_contracted_extension(extension: str) → str[source]¶

Returns the expanded version of a known contracted extension.

This function maps extensions like .tgz to .tar.gz. On unknown extensions, return the input unmodified.

spack.util.url.expand_contracted_extension_in_path(path_or_url: str, *, extension: str | None = None) → str[source]¶

Returns the input path or URL with any contraction extension expanded.

Parameters:

path_or_url – path or URL to be expanded
extension – if specified, only attempt to expand that extension

spack.util.url.extension_from_path(path_or_url: str | None) → str | None[source]¶

Tries to match an allowed archive extension to the input. Returns the first match, or None if no match was found.

Raises:: ValueError – if the input is None

spack.util.url.file_url_string_to_path(url)[source]¶

spack.util.url.find_list_urls(url: str) → Set[str][source]¶

Find good list URLs for the supplied URL.

By default, returns the dirname of the archive path.

Provides special treatment for the following websites, which have a unique list URL different from the dirname of the download URL:

GitHub	`https://github.com/<repo>/<name>/releases`
GitLab	`https://gitlab.\*/<repo>/<name>/tags`
BitBucket	`https://bitbucket.org/<repo>/<name>/downloads/?tab=tags`
CRAN	`https://\*.r-project.org/src/contrib/Archive/<name>`
PyPI	`https://pypi.org/simple/<name>/`
LuaRocks	`https://luarocks.org/modules/<repo>/<name>`

Note: this function is called by spack versions, spack checksum, and spack create, but not by spack fetch or spack install.

Parameters:: url (str) – The download URL for the package
Returns:: One or more list URLs for the package
Return type:: set

spack.util.url.format(parsed_url)[source]¶

Format a URL string

Returns a canonicalized format of the given URL as a string.

spack.util.url.has_extension(path_or_url: str, ext: str) → bool[source]¶: Returns true if the extension in input is present in path, false otherwise.

spack.util.url.is_path_instead_of_url(path_or_url)[source]¶: Historically some config files and spack commands used paths where urls should be used. This utility can be used to validate and promote paths to urls.

spack.util.url.join(base: str, *components: str, resolve_href: bool = False, **kwargs) → str[source]¶

Convenience wrapper around urllib.parse.urljoin(), with a few differences:

By default resolve_href=False, which makes the function like os.path.join(). For example https://example.com/a/b + c/d = https://example.com/a/b/c/d. If resolve_href=True, the behavior is how a browser would resolve the URL: https://example.com/a/c/d.
s3://, gs://, oci:// URLs are joined like http:// URLs.
It accepts multiple components for convenience. Note that components[1:] are treated as literal path components and appended to components[0] separated by slashes.

spack.util.url.local_file_path(url)[source]¶

Get a local file path from a url.

If url is a file:// URL, return the absolute path to the local file or directory referenced by it. Otherwise, return None.

spack.util.url.parse_link_rel_next(link_value: str) → str | None[source]¶: Return the next link from a Link header value, if any.

spack.util.url.path_to_file_url(path)[source]¶

spack.util.url.remove_extension(path_or_url: str, *, extension: str) → str[source]¶: Returns the input with the extension removed

spack.util.url.split_url_extension(url: str) → Tuple[str, ...][source]¶

Some URLs have a query string, e.g.:

https://github.com/losalamos/CLAMR/blob/packages/PowerParser_v2.0.7.tgz?raw=true
http://www.apache.org/dyn/closer.cgi?path=/cassandra/1.2.0/apache-cassandra-1.2.0-rc2-bin.tar.gz
https://gitlab.kitware.com/vtk/vtk/repository/archive.tar.bz2?ref=v7.0.0

In (1), the query string needs to be stripped to get at the extension, but in (2) & (3), the filename is IN a single final query argument.

This strips the URL into three pieces: prefix, ext, and suffix. The suffix contains anything that was stripped off the URL to get at the file extension. In (1), it will be '?raw=true', but in (2), it will be empty. In (3) the suffix is a parameter that follows after the file extension, e.g.:

('https://github.com/losalamos/CLAMR/blob/packages/PowerParser_v2.0.7', '.tgz', '?raw=true')
('http://www.apache.org/dyn/closer.cgi?path=/cassandra/1.2.0/apache-cassandra-1.2.0-rc2-bin', '.tar.gz', None)
('https://gitlab.kitware.com/vtk/vtk/repository/archive', '.tar.bz2', '?ref=v7.0.0')

spack.util.url.split_url_on_sourceforge_suffix(url: str) → Tuple[str, ...][source]¶: If the input is a sourceforge URL, returns base URL and /download suffix. Otherwise, returns the input URL and an empty string.

spack.util.url.strip_compression_extension(path_or_url: str, ext: str | None = None) → str[source]¶

Strips the compression extension from the input, and returns it. For instance, "foo.tgz" becomes "foo.tar".

If no extension is given, try a default list of extensions.

Parameters:

path_or_url – input to be stripped
ext – if given, extension to be stripped

spack.util.url.strip_extension(path_or_url: str, *, extension: str | None = None) → str[source]¶

If a path contains the extension in input, returns the path stripped of the extension. Otherwise, returns the input path.

If extension is None, attempts to strip any allowed extension from path.

spack.util.url.strip_query_and_fragment(url: str) → Tuple[str, str][source]¶

Strips query and fragment from a url, then returns the base url and the suffix.

Parameters:: url – URL to be stripped
Raises:: ValueError – when there is any error parsing the URL

spack.util.url.strip_version_suffixes(path_or_url: str) → str[source]¶

Some tarballs contain extraneous information after the version:

bowtie2-2.2.5-source
libevent-2.0.21-stable
cuda_8.0.44_linux.run

These strings are not part of the version number and should be ignored. This function strips those suffixes off and returns the remaining string. The goal is that the version is always the last thing in path:

bowtie2-2.2.5
libevent-2.0.21
cuda_8.0.44

Parameters:: path_or_url – The filename or URL for the package
Returns:: The path with any extraneous suffixes removed

spack.util.url.validate_scheme(scheme)[source]¶: Returns true if the URL scheme is generally known to Spack. This function helps mostly in validation of paths vs urls, as Windows paths such as C:/x/y/z (with backward not forward slash) may parse as a URL with scheme C and path /x/y/z.

spack.util.web module¶

exception spack.util.web.DetailedHTTPError(req: Request, code: int, msg: str, hdrs: Message, fp: IO | None)[source]¶: Bases: HTTPError

exception spack.util.web.DetailedURLError(req: Request, reason)[source]¶: Bases: URLError

class spack.util.web.ExtractMetadataParser[source]¶

Bases: HTMLParser

This parser takes an HTML page and selects the include-fragments, used on GitHub, https://github.github.io/include-fragment-element, as well as a possible base url.

handle_starttag(tag, attrs)[source]¶

exception spack.util.web.HTMLParseError[source]¶: Bases: Exception

class spack.util.web.LinkParser[source]¶

Bases: HTMLParser

This parser just takes an HTML page and strips out the hrefs on the links, as well as some javascript tags used on GitLab servers. Good enough for a really simple spider.

handle_starttag(tag, attrs)[source]¶

exception spack.util.web.NoNetworkConnectionError(message, url)[source]¶

Bases: SpackWebError

Raised when an operation can’t get an internet connection.

class spack.util.web.Retry(total: int = 5, backoff_factor: float = 1.0, backoff_jitter: float = 0.0, backoff_max: float = 120.0)[source]¶

Bases: object

Wrapper class around retry logic

backoff() → float[source]¶: Return the backoff duration in seconds for the current attempt

is_exhausted()[source]¶: Return if this the retry counter is exhausted

is_last_attempt()[source]¶: Return if this the retry counter is on last attempt

sleep() → None[source]¶: Sleep for the backoff duration of the current attempt

spack.util.web.SPACK_USER_AGENT¶: User-Agent used in Request objects

class spack.util.web.SpackHTTPDefaultErrorHandler[source]¶

Bases: HTTPDefaultErrorHandler

http_error_default(req, fp, code, msg, hdrs)[source]¶

class spack.util.web.SpackHTTPSHandler(debuglevel=None, context=None, check_hostname=None)[source]¶

Bases: HTTPSHandler

A custom HTTPS handler that shows more detailed error messages on connection failure.

https_open(req)[source]¶

exception spack.util.web.SpackWebError(message: str, long_message: str | None = None)[source]¶

Bases: SpackError

Superclass for Spack web spidering errors.

spack.util.web.base_curl_fetch_args(url, timeout=0)[source]¶

Return the basic fetch arguments typically used in calls to curl.

The arguments include those for ensuring behaviors such as failing on errors for codes over 400, printing HTML headers, resolving 3xx redirects, status or failure handling, and connection timeouts.

It also uses the following configuration option to set an additional argument as needed:

config:connect_timeout (int): connection timeout
config:verify_ssl (str): Perform SSL verification

Parameters:

url (str) – URL whose contents will be fetched
timeout (int) – Connection timeout, which is only used if higher than config:connect_timeout

Returns (list): list of argument strings

spack.util.web.check_curl_code(returncode: int) → None[source]¶

Check standard return code failures for provided arguments.

Parameters:: returncode – curl return code

Raises FetchError if the curl returncode indicates failure

spack.util.web.custom_ssl_certs() → Tuple[bool, str] | None[source]¶: Returns a tuple (is_file, path) if custom SSL certifates are configured and valid.

spack.util.web.fetch_url_text(url, curl: Executable | None = None, dest_dir='.')[source]¶

Retrieves text-only URL content using the configured fetch method. It determines the fetch method from:

config:url_fetch_method (str): fetch method to use (e.g., ‘curl’)

If the method is curl, it also uses the following configuration options:

config:connect_timeout (int): connection time out
config:verify_ssl (str): Perform SSL verification

Parameters:

url (str) – URL whose contents are to be fetched
curl (spack.util.executable.Executable or None) – (optional) curl executable if curl is the configured fetch method
dest_dir (str) – (optional) destination directory for fetched text file

Returns (str or None): path to the fetched file

Raises FetchError if the curl returncode indicates failure

spack.util.web.get_header(headers, header_name)[source]¶

Looks up a dict of headers for the given header value.

Looks up a dict of headers, [headers], for a header value given by [header_name]. Returns headers[header_name] if header_name is in headers. Otherwise, the first fuzzy match is returned, if any.

This fuzzy matching is performed by discarding word separators and capitalization, so that for example, “Content-length”, “content_length”, “conTENtLength”, etc., all match. In the case of multiple fuzzy-matches, the returned value is the “first” such match given the underlying mapping’s ordering, or unspecified if no such ordering is defined.

If header_name is not in headers, and no such fuzzy match exists, then a KeyError is raised.

spack.util.web.is_transient_error(e: Exception) → bool[source]¶: Return True for HTTP/network errors that are worth retrying.

spack.util.web.list_url(url, recursive=False)[source]¶

spack.util.web.parse_etag(header_value)[source]¶: Parse a strong etag from an ETag: <value> header value. We don’t allow for weakness indicators because it’s unclear what that means for cache invalidation.

spack.util.web.push_to_url(local_file_path, remote_path, keep_original=True, extra_args=None)[source]¶

spack.util.web.read_from_url(url, accept_content_type=None)[source]¶

spack.util.web.read_json(url: str)[source]¶: Fetch url and return the response body parsed as JSON.

spack.util.web.read_text(url: str) → str[source]¶: Fetch url and return the response body decoded as UTF-8 text.

spack.util.web.remove_url(url, recursive=False)[source]¶

spack.util.web.require_curl() → Executable[source]¶

spack.util.web.retry_on_transient_error(f: Callable[[_P], _R], retry: Retry | None = None) → Callable[[_P], _R][source]¶: Retry a function on transient HTTP/network errors with exponential backoff.

spack.util.web.set_curl_env_for_ssl_certs(curl: Executable) → None[source]¶: configure curl to use custom certs in a file at runtime. See: https://curl.se/docs/sslcerts.html item 4

spack.util.web.spider(root_urls: str | Iterable[str], depth: int = 0, concurrency: int | None = None)[source]¶

Get web pages from root URLs.

If depth is specified (e.g., depth=2), then this will also follow up to <depth> levels of links from each root.

Parameters:

root_urls – root urls used as a starting point for spidering
depth – level of recursion into links
concurrency – number of simultaneous requests that can be sent

Returns:

A dict of pages visited (URL) mapped to their full text and the set of visited links.

spack.util.web.ssl_create_default_context()[source]¶: Create the default SSL context for urllib with custom certificates if configured.

spack.util.web.stat_url(url: str) → Tuple[int, float] | None[source]¶

Get stat result for a URL.

Parameters:: url – URL to get stat result for
Returns:: A tuple of (size, mtime) if the URL exists, None otherwise.

spack.util.web.url_exists(url, curl=None)[source]¶

Determines whether url exists.

A scheme-specific process is used for Google Storage (gs) and Amazon Simple Storage Service (s3) URLs; otherwise, the configured fetch method defined by config:url_fetch_method is used.

Parameters:

url (str) – URL whose existence is being checked
curl (spack.util.executable.Executable or None) – (optional) curl executable if curl is the configured fetch method

Returns (bool): True if it exists; False otherwise.

spack.util.web.urlopen¶: Dispatches to the correct OpenerDirector.open, based on Spack configuration.

spack.util.windows_registry module¶

Utility module for dealing with Windows Registry.

class spack.util.windows_registry.HKEY[source]¶

Bases: object

Predefined, open registry HKEYs From the Microsoft docs: An application must open a key before it can read data from the registry. To open a key, an application must supply a handle to another key in the registry that is already open. The system defines predefined keys that are always open. Predefined keys help an application navigate in the registry.

HKEY_CLASSES_ROOT¶

HKEY_CURRENT_CONFIG¶

HKEY_CURRENT_USER¶

HKEY_LOCAL_MACHINE¶

HKEY_PERFORMANCE_DATA¶

HKEY_USERS¶

exception spack.util.windows_registry.InvalidKeyError(key)[source]¶

Bases: RegistryError

Runtime Error describing issue with invalid key access to Windows registry

exception spack.util.windows_registry.InvalidRegistryOperation(name, e, *args, **kwargs)[source]¶

Bases: RegistryError

A Runtime Error encountered when a registry operation is invalid for an indeterminate reason

exception spack.util.windows_registry.RegistryError[source]¶

Bases: Exception

RunTime Error concerning the Windows Registry

class spack.util.windows_registry.RegistryKey(name, handle)[source]¶

Bases: object

Class wrapping a Windows registry key

EnumKey(index)[source]¶: Convenience wrapper around winreg.EnumKey

EnumValue(index)[source]¶: Convenience wrapper around winreg.EnumValue

OpenKeyEx(subname, **kwargs)[source]¶: Convenience wrapper around winreg.OpenKeyEx

QueryInfoKey()[source]¶: Convenience wrapper around winreg.QueryInfoKey

QueryValueEx(name, **kwargs)[source]¶: Convenience wrapper around winreg.QueryValueEx

get_subkey(sub_key)[source]¶: Returns subkey of name sub_key in a RegistryKey objects

get_value(val_name)[source]¶: Returns value associated with this key in RegistryValue object

property hkey¶

property subkeys¶: Returns list of all subkeys of this key as RegistryKey objects

property values¶: Returns all subvalues of this key as RegistryValue objects in dictionary of value name : RegistryValue object

winreg_error_handler(name, *args, **kwargs)[source]¶

class spack.util.windows_registry.RegistryValue(name, value, parent_key)[source]¶

Bases: object

Class defining a Windows registry entry

class spack.util.windows_registry.WindowsRegistryView(key, root_key=HKEY.HKEY_CURRENT_USER)[source]¶

Bases: object

Interface to provide access, querying, and searching to Windows registry entries. This class represents a single key entrypoint into the Windows registry and provides an interface to this key’s values, its subkeys, and those subkey’s values. This class cannot be used to move freely about the registry, only subkeys/values of the root key used to instantiate this class.

class KeyMatchConditions[source]¶

Bases: object

static name_matcher(subkey_name)[source]¶

static regex_matcher(subkey_name)[source]¶

find_matching_subkey(subkey_name: str, recursive: bool = True)[source]¶

Perform a BFS of subkeys until a key matching subkey name regex is found Returns None or the first RegistryKey object corresponding to requested key name

Parameters:

subkey_name – subkey to be searched for
recursive – perform a recursive search

Returns:

the desired subkey as a RegistryKey object, or none

find_subkey(subkey_name: str, recursive: bool = True)[source]¶

Perform a BFS of subkeys until desired key is found Returns None or RegistryKey object corresponding to requested key name

Parameters:

subkey_name – subkey to be searched for
recursive – perform a recursive search

Returns:

the desired subkey as a RegistryKey object, or none

find_subkeys(subkey_name: str, recursive: bool = True)[source]¶

Exactly the same as find_subkey, except this function tries to match a regex to multiple keys

Parameters:: subkey_name – subkey to be searched for
Returns:: the desired subkeys as a list of RegistryKey object, or none

find_value(val_name: str, recursive: bool = True)[source]¶

If non recursive, return RegistryValue object corresponding to name

Parameters:

val_name – name of value desired from registry
recursive – optional argument, if True, the registry is searched recursively for the value of name val_name, else only the current key is searched

Returns:

The desired registry value as a RegistryValue object if it exists, otherwise, None

get_matching_subkeys(subkey_name)[source]¶

Returns all subkeys regex matching subkey name

Note: this method obtains only direct subkeys of the given key and does not descend to transitive subkeys. For this behavior, see find_matching_subkeys

get_subkey(subkey_name)[source]¶

get_subkeys()[source]¶

get_value(value_name)[source]¶: Return registry value corresponding to provided argument (if it exists)

get_values()[source]¶

invalid_reg_ref_error_handler()[source]¶

property reg¶