minitrino.utils module#

Utility functions for Minitrino CLI and core operations.

minitrino.utils.pass_environment() → Any[source][source]#

Return a Click pass decorator for the MinitrinoContext.

Returns:: A decorator that passes the MinitrinoContext instance.
Return type:: Any

minitrino.utils.handle_exception(error: BaseException, ctx: Any | None = None, additional_msg: str = '', skip_traceback: bool = False) → None[source][source]#

Handle a single exception.

Parameters:

error (BaseException) – The exception object.
ctx (Optional[Any]) – Optional CLI context object with logger.
additional_msg (str) – Additional message to log, if any.
skip_traceback (bool) – If True, suppresses traceback output unless overridden by error type.

Raises:

SystemExit – Exits the program with the appropriate exit code.

minitrino.utils.exception_handler(func: Any) → Any[source][source]#

Handle unhandled exceptions.

Parameters:: func (Callable) – The function to wrap.
Returns:: The wrapped function with exception handling.
Return type:: Callable

minitrino.utils.check_daemon(docker_client: Any) → None[source][source]#

Check if the Docker daemon is running.

Parameters:: docker_client (Any) – Docker client instance.
Raises:: UserError – If the Docker daemon is not running or cannot be pinged.

minitrino.utils.check_lib(ctx: MinitrinoContext) → None[source][source]#

Check if a Minitrino library exists.

Parameters:: ctx (MinitrinoContext) – Context object containing library directory information.

minitrino.utils.container_user_and_id(ctx: MinitrinoContext | None = None, container: Container | MinitrinoContainer | str = '') → tuple[str, str][source][source]#

Return the build user and build user ID for a cluster container.

Parameters:

ctx (MinitrinoContext) – Context object containing cluster information.
container (Container | MinitrinoContainer | str) – Container object or container name.

Returns:

Tuple of build user and build user ID.

Return type:

tuple[str, str]

Raises:

ValueError – If container is not provided.

Notes

Commands executed in coordinator/worker containers tend to rely on environment variables set during the build process. This function returns the build user and build user ID for a cluster container, which can then be used to execute commands in the container with the correct UID to ensure environment variables resolve correctly.

Examples

>>> _, uid = container_user_and_id("minintrino-default")
>>> cmd = ["cat /etc/${CLUSTER_DIST}/config.properties"]
>>> cmd_executor.execute(cmd, container="minintrino-default", user=uid)

minitrino.utils.generate_identifier(identifiers: dict[str, Any] | None = None) → str[source][source]#

Return an object identifier string used for creating log messages.

Parameters:: identifiers (Optional[Dict[str, Any]], optional) – Dictionary of “identifier_key”: “identifier_value” pairs, by default None.
Returns:: Formatted string with identifiers enclosed in brackets.
Return type:: str

Examples

>>> generate_identifier({"cluster": "default", "module": "test"})
>>> '[cluster: default] [module: test]'

minitrino.utils.parse_key_value_pair(ctx: MinitrinoContext, pair: str, hard_fail: bool = False) → tuple[str, str][source][source]#

Parse a key-value pair from a string.

Parameters:

pair (str) – Key-value pair to parse.
hard_fail (bool, optional) – Whether to raise an error if the key-value pair is invalid, by default False.

Returns:

Tuple of key and value.

Return type:

tuple[str, str]

minitrino.utils.closest_match_or_error(name: str, valid_names: list[str], context: str = 'item') → str[source][source]#

Return the name or fail with a closest match suggestion.

Parameters:

name (str) – The user-provided name to validate.
valid_names (list[str]) – List of valid names to check against.
context (str, optional) – Context string for error message (default: “item”).

Returns:

The valid name (if found).

Return type:

str

Raises:

UserError – If the name is not valid, with a suggestion if available.

Examples

>>> closest_match_or_error('ressources', ['resources', 'remove'])
UserError: Item 'ressources' not found. Did you mean 'resources'?

minitrino.utils.validate_yes(value: str) → bool[source][source]#

Validate if the input is an affirmative response.

Parameters:: value (str) – Value to validate.
Returns:: True if the input is ‘y’ or ‘yes’ (case-insensitive), False otherwise.
Return type:: bool

minitrino.utils.cli_ver() → str[source][source]#

Return the CLI version.

Returns:: CLI version.
Return type:: str

minitrino.utils.lib_ver(ctx: MinitrinoContext | None = None, lib_path: str = '') → str[source][source]#

Return the library version.

Returns:: Library version.
Return type:: str