Skip to content

Utilities

consortium.framework.utils

ContainerBuildError(message='', detail=None)

Bases: ComponentRuntimeError

Raised when a containerized build fails.

Subclasses ComponentRuntimeError, so raising it from an agent generator build step, a listener, or a plugin surfaces as that component's own runtime error without the caller needing to catch and re-raise it.

ContainerRuntimeUnavailableError(message='', detail=None)

Bases: ComponentStartError

Raised when no usable container runtime can be reached.

Subclasses ComponentStartError, so raising it from a component's start hook aborts the start with a clean, client-facing error rather than an unhandled traceback.

chunked(items, size)

Iterate over a sequence in fixed-size groups.

Parameters:

Name Type Description Default
items Sequence[T]

Sequence to split into groups.

required
size int

Maximum number of items in each group.

required

Yields:

Type Description
Sequence[T]

Consecutive slices of the input sequence.

Raises:

Type Description
ValueError

If size is less than one.

dedupe_preserving_order(items)

Return unique items in their first-seen order.

Parameters:

Name Type Description Default
items Iterable[HashableT]

Iterable containing hashable values.

required

Returns:

Type Description
list[HashableT]

A list containing each distinct input value once.

build_artifact_in_container(*, image_tag, command, artifact_path, output_directory, output_name=None, environment=None) async

Run a build command in a throwaway container and retrieve the file it produces.

The artifact is copied out of the container rather than shared through a mounted directory. Mounts are resolved by the engine, which is not necessarily the machine this code runs on: when the server itself runs in a container against a mounted docker socket, a mount of a local path silently resolves to an empty directory on the engine's filesystem. Copying keeps the build identical whether the engine is local, remote, or the host's engine seen from inside a container.

The container is always removed, including when the build command fails or the surrounding task is cancelled.

Parameters:

Name Type Description Default
image_tag str

Image to run, as returned by build_container_image.

required
command Sequence[str]

Command and arguments to execute inside the container.

required
artifact_path str | PurePosixPath

Absolute path of the file to retrieve, inside the container.

required
output_directory str | Path

Directory on this machine to place the retrieved file in.

required
output_name str | None

Filename to save the artifact under. Defaults to a per-build name, so that concurrent builds cannot overwrite each other's output.

None
environment Mapping[str, str] | None

Environment variables to set inside the container. This is the supported way to vary a build (compiler targets, feature flags), since the same image is reused across builds.

None

Returns:

Type Description
Path

Path of the retrieved artifact. The caller owns the file and is responsible

Path

for moving or deleting it.

Raises:

Type Description
ContainerBuildError

If the build command fails, or the artifact cannot be retrieved from the container.

build_container_image(context_directory, *, dockerfile=None, image_tag=None) async

Build a container image from a directory and return its tag.

Parameters:

Name Type Description Default
context_directory str | Path

Directory sent to the engine as the build context. Its Dockerfile is used unless dockerfile is given.

required
dockerfile str | Path | None

Dockerfile to build with, when it does not sit at the root of the build context.

None
image_tag str | None

Tag to apply. Defaults to a stable tag derived from the context directory, which keeps the layer cache warm across builds and cannot collide with another component's image.

None

Returns:

Type Description
str

The tag the image was built under, to be passed to

str

build_artifact_in_container.

Raises:

Type Description
ContainerBuildError

If the image fails to build.

ensure_container_runtime_available() async

Verify that a container runtime is installed and reachable.

Call this from a component's start hook so the component refuses to start on a machine that cannot build, instead of failing partway through a build.

Raises:

Type Description
ContainerRuntimeUnavailableError

If the docker client is not on the system path, or no engine is currently reachable through it.

find_available_port(address='127.0.0.1', start=49152, end=65535)

Find the first available port in an inclusive range.

Parameters:

Name Type Description Default
address str

Local address against which to test each port.

'127.0.0.1'
start int

First port in the search range.

49152
end int

Last port in the search range.

65535

Returns:

Type Description
int | None

The first bindable port, or None when no port is available.

Raises:

Type Description
ValueError

If the range is reversed or either boundary is invalid.

get_local_ip(fallback='127.0.0.1')

Return the local address selected for an outbound connection.

Parameters:

Name Type Description Default
fallback str

Address to return when the local address cannot be determined.

'127.0.0.1'

Returns:

Type Description
str

The selected local IP address or the fallback address.

get_network_interfaces()

Return discovered network-interface names and IPv4 addresses.

Returns:

Type Description
list[tuple[str, str]]

Interface-name and IP-address pairs discovered from the operating system.

get_public_ip(*, timeout_seconds=5.0) async

Resolve the public IP address reported by the external IP service.

Parameters:

Name Type Description Default
timeout_seconds float

Maximum time to wait for the external request.

5.0

Returns:

Type Description
str | None

The reported public IP address, or None when it cannot be resolved.

is_bindable(address, port)

Check whether a TCP socket can currently bind to an address and port.

Parameters:

Name Type Description Default
address str

Local address to bind.

required
port int

Port to bind.

required

Returns:

Type Description
bool

True when the socket can be bound, otherwise False.

is_valid_ip(value)

Check whether a value is a valid IPv4 or IPv6 address.

Parameters:

Name Type Description Default
value str

Address string to validate.

required

Returns:

Type Description
bool

True when the value is a valid IP address, otherwise False.

is_valid_port(port)

Check whether a port number can be used for a network socket.

Parameters:

Name Type Description Default
port int

Port number to validate.

required

Returns:

Type Description
bool

True when the port is in the range 1 through 65535, otherwise False.

ensure_within_root(path, root)

Resolve a path and verify that it is contained by a root directory.

Parameters:

Name Type Description Default
path str | PathLike[str]

Path to resolve and validate.

required
root str | PathLike[str]

Directory that must contain the resolved path.

required

Returns:

Type Description
Path

The resolved path.

Raises:

Type Description
ValueError

If the resolved path is outside the resolved root directory.

unique_path(path)

Return a path that does not currently exist.

Parameters:

Name Type Description Default
path str | PathLike[str]

Preferred path to use.

required

Returns:

Type Description
Path

The preferred path when available, otherwise a suffixed alternative path.

random_string(character_set, length=16)

Return a cryptographically secure string from a supplied character set.

Parameters:

Name Type Description Default
character_set str

Characters from which to sample each output character.

required
length int

Number of characters to generate.

16

Returns:

Type Description
str

A string containing characters sampled from the supplied character set.

Raises:

Type Description
ValueError

If character_set is empty or length is negative.

replace_all(value, replacements)

Replace multiple substrings in a string in mapping iteration order.

Parameters:

Name Type Description Default
value str

Original string in which to replace substrings.

required
replacements Mapping[str, str]

Substrings and their replacement values.

required

Returns:

Type Description
str

The string after all replacements have been applied.