opsicommon.utils package

Submodules

opsicommon.utils.rsync module

rsync

class opsicommon.utils.rsync.Buffer

Bases: Structure

avail_in

Structure/Union member

avail_out

Structure/Union member

eof_in

Structure/Union member

next_in

Structure/Union member

next_out

Structure/Union member

exception opsicommon.utils.rsync.LibrsyncError(res)

Bases: Exception

Parameters:

res (int)

opsicommon.utils.rsync.librsync_delta_file(filename, signature, deltafile)

Create delta file from original file and the signature of the file to patch.

Parameters:

  • filename (Path | str)

  • signature (bytes)

  • deltafile (Path | str)

Return type:

None

opsicommon.utils.rsync.librsync_patch_file(oldfile, deltafile, newfile)

Create the new file from old file and delta file.

Parameters:

  • oldfile (Path | str)

  • deltafile (Path | str)

  • newfile (Path | str)

Return type:

None

opsicommon.utils.rsync.librsync_signature(filename, base64_encoded=True)

Get the signature of the file to patch.

Parameters:

  • filename (Path | str)

  • base64_encoded (bool)

Return type:

str | bytes

Module contents

General utility functions.

class opsicommon.utils.LegacyVersion(version)

Bases: Version

Parameters:

version (str)

class opsicommon.utils.Singleton

Bases: type

opsicommon.utils.combine_versions(obj)

Returns the combination of product and package version.

Returns:

The version.

Return type:

str

Parameters:

obj (Union[Product, ProductOnClient, ProductOnDepot])

opsicommon.utils.compare_versions(version1, condition, version2)

Compare the versions v1 and v2 with the given condition.

condition may be one of ==, =, <, <=, >, >=.

Raises:

ValueError – If invalid value for version or condition if given.

Return type:

bool

Returns:

If the comparison matches this will return True.

Parameters:

  • version1 (str)

  • condition (Literal['==', '=', '<', '<=', '>', '>='])

  • version2 (str)

opsicommon.utils.compress_data(data, compression, compression_level=0, lz4_block_linked=True)
Parameters:

  • data (bytes)

  • compression (Literal['lz4', 'deflate', 'gz', 'gzip'])

  • compression_level (int)

  • lz4_block_linked (bool)

Return type:

bytes

opsicommon.utils.decompress_data(data, compression)
Parameters:

  • data (bytes)

  • compression (Literal['lz4', 'deflate', 'gz', 'gzip'])

Return type:

bytes

opsicommon.utils.execute(cmd, allow_exit_codes=None)
Parameters:

  • cmd (list[str])

  • allow_exit_codes (list[int | EllipsisType] | tuple[int | EllipsisType] | None)

Return type:

CompletedProcess

opsicommon.utils.frozen_lru_cache(*decorator_args)

This decorator is intended to be used as drop-in replacement for functools.lru_cache. It mitigates the weakness of not being able to handle dictionary type arguments by freezing them.

Parameters:

decorator_args (Any)

Return type:

Callable

opsicommon.utils.generate_opsi_host_key()

Generates an random opsi host key.

On Python 3.5 or lower this will try to make use of an existing random device. As a fallback the generation is done in plain Python.

Return type:

str

opsicommon.utils.ip_address_in_network(address, network)

Checks if the given IP address is in the given network range. Returns True if the given address is part of the network. Returns False if the given address is not part of the network.

Parameters:

  • address (str) – The IP which we check.

  • network (str) – The network address written with slash notation.

Return type:

bool

opsicommon.utils.make_temp_dir(base=None)
Parameters:

base (Path | None)

Return type:

Generator[Path, None, None]

opsicommon.utils.monkeypatch_subprocess_for_frozen()
Return type:

None

opsicommon.utils.prepare_proxy_environment(hostname, proxy_url='system', no_proxy_addresses=None, session=None)

proxy_url can be: * an explicid url like http://10.10.10.1:8080 * the string “system” in which case the os environment determines proxy behaviour * emptystring or None to disable proxy usage. If session is given its proxy settings are adapted. Else a new session is created and returned.

Parameters:

  • hostname (str)

  • proxy_url (str | None)

  • no_proxy_addresses (list[str] | None)

  • session (Session | None)

Return type:

Session

opsicommon.utils.retry(retries=3, wait=0, exceptions=None, caught_exceptions=None)

Decorator to retry a function. :type retries: int :param retries: Number of retries :type wait: float :param wait: Time to wait between retries :type exceptions: Optional[Iterable[Type[Exception]]] :param exceptions: Exception to catch, if None catch all exceptions

Parameters:

caught_exceptions (list[Exception] | None)

Return type:

Callable

opsicommon.utils.timestamp(secs=0.0, date_only=False)

Returns a timestamp of the current system time in format YYYY-mm-dd[ HH:MM:SS]

Parameters:

  • secs (float)

  • date_only (bool)

Return type:

str

opsicommon.utils.unix_timestamp(*, millis=False, add_seconds=0.0)

Returns the current unix timestamp (UTC). If millis is True, the timestamp is in milliseconds. add_seconds can be used to add or subtract seconds from the current time.

Parameters:

  • millis (bool)

  • add_seconds (float)

Return type:

float

opsicommon.utils.update_environment_from_config_files(files=None)

Updates the environment variables from the config files.

Parameters:

files (list[Path] | None)

Return type:

None

opsicommon.utils.utc_timestamp(date_only=False)

Returns a UTC timestamp in format YYYY-mm-dd[ HH:MM:SS]

Parameters:

date_only (bool)

Return type:

str