Utilities¶

The utilities module provides helper functions and common utilities used throughout the CLI for various tasks such as display formatting, logging, version management, and more.

General Utilities¶

General-purpose utility functions used across the SDK.

Licensed under the Kelvin Inc. Developer SDK License Agreement (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.kelvininc.com/developer-sdk-license

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

kelvin.sdk.lib.utils.general_utils.guess_delimiter(line, delimiters=('\\t', '\\x00', '|', ','))[source]¶

Guess delimiter in line.

Return type:

str

Parameters:

line (str)
delimiters (Iterable[str])

Parameters¶

line: str: line of delimited text.
delimiters: Iterable[str]: delimiters to check.

Returns¶

str:: the guessed delimiter string.

kelvin.sdk.lib.utils.general_utils.dict_to_yaml(content, comment_header=None)[source]¶

Convert a dictionary to a yaml.

Return type:

str

Parameters:

content (dict)
comment_header (Any | None)

Parameters¶

content: dict: the dictionary content to convert to the yaml format.
comment_header: Optional[Any]: the comment header to include in the yaml file. Do not include the ‘#’ character.

Returns¶

str:: A string with the yaml format content.

kelvin.sdk.lib.utils.general_utils.get_url_encoded_string(original_string)[source]¶

Return the url-encoded version of the provided string.

Return type:: str
Parameters:: original_string (str)

Parameters¶

original_stringstr: the string to url encode

Returns¶

str:: a url encoded string

kelvin.sdk.lib.utils.general_utils.standardize_string(value)[source]¶

Given a specific value, replace its spaces and dashes with underscores to be snake-case compliant.

Return type:: str
Parameters:: value (str)

Parameters¶

value: str: the string to be ‘standardized’.

Returns¶

str:: the new, standardized string.

kelvin.sdk.lib.utils.general_utils.camel_name(name)[source]¶

Create camel-case name from name.

Return type:: str
Parameters:: name (str)

kelvin.sdk.lib.utils.general_utils.open_link_in_browser(link)[source]¶

Open the specified link on the default web browser.

Return type:: bool
Parameters:: link (str)

Parameters¶

link: str: the link to open

Returns¶

bool:: a boolean indicating whether the link was successfully opened.

kelvin.sdk.lib.utils.general_utils.get_requirements_from_file(file_path)[source]¶

When provided with a path to a requirements file, yield a list of its requirements.

Return type:: List[str]
Parameters:: file_path (Path | None)

Parameters¶

file_pathPath: the Path to the desired requirements file

Returns¶

List[str]: a list containing all requirements in the file

kelvin.sdk.lib.utils.general_utils.merge(x, *args, **kwargs)[source]¶

Merge two dictionaries.

Return type:

TypeVar(T, bound= MutableMapping[str, Any])

Parameters:

x (T)
args (Mapping[str, Any] | None)
kwargs (Any)

Parameters¶

xdict: the initial, mutable dictionary.
argsMapping[str, Any]: the arguments to merge into the ‘x’ dictionary.
kwargsAny: the keyword arguments to merge into the ‘x’ dictionary.

Returns¶

dictionary:: the initial, mutated X dictionary.

kelvin.sdk.lib.utils.general_utils.flatten(x)[source]¶

Flatten nested mappings.

Return type:: Sequence[Tuple[str, Any]]
Parameters:: x (Mapping[str, Any])

kelvin.sdk.lib.utils.general_utils.inflate(x, separator='.', result=None)[source]¶

Inflate flattened keys via separator into nested dictionary.

Return type:

Dict[str, Any]

Parameters:

x (Mapping[str, Any])
separator (str)
result (Dict[str, Any] | None)

kelvin.sdk.lib.utils.general_utils.get_bytes_as_human_readable(input_bytes_data)[source]¶

When provided with bytes data, return its ‘human-readable’ version.

Return type:: str
Parameters:: input_bytes_data (int | None | float)

Parameters¶

input_bytes_dataUnion[Optional[int], Optional[float]]: the input int that corresponds to the bytes value.

Returns¶

str:: a string containing the human-readable version of the input.

kelvin.sdk.lib.utils.general_utils.get_iec_to_si_format_as_human_readable(input_data)[source]¶

When provided with a str in iec format, return its ‘human-readable’ version. ex: 256Mi

Return type:: str
Parameters:: input_data (str)

Parameters¶

input_datastr: the input str in iec format. ex: 256Mi

Returns¶

str:: a string containing the human-readable version of the input.

kelvin.sdk.lib.utils.general_utils.get_datetime_as_human_readable(input_date)[source]¶

When provided with a datetime, retrieve its human readable form with the base date and its difference.

Return type:: str
Parameters:: input_date (float | datetime | None)

Parameters¶

input_dateUnion[float, Optional[datetime]]: the datetime to display.

Returns¶

str:: a string containing both the human readable datetime plus the difference to ‘now’

kelvin.sdk.lib.utils.general_utils.parse_pydantic_errors(validation_error)[source]¶

Parse the provided ValidationError and break it down to a ‘pretty’ string message.

Return type:: str
Parameters:: validation_error (ValidationError)

Parameters¶

validation_errorValidationError: the ValidationError to prettify.

Returns¶

str:: a ‘pretty’ string with the parsed errors.

kelvin.sdk.lib.utils.general_utils.get_files_from_dir(file_type, input_dir)[source]¶

Retrieve all files of a given type from the specified directory.

Return type:

List

Parameters:

file_type (str)
input_dir (str)

Parameters¶

file_typestr: the file type to search for.
input_dirstr: the directory to read the files from.

Returns¶

List:: the list of all matching files

kelvin.sdk.lib.utils.general_utils.unique_items(items)[source]¶

When provided with a list of items, retrieve the same list without duplicates and with the same order.

Return type:: List
Parameters:: items (List)

Parameters¶

itemsList: the original list.

Returns¶

List:: the ordered list.

kelvin.sdk.lib.utils.general_utils.lower(x)[source]¶

Lower representation of data for serialisation.

Return type:: Any
Parameters:: x (Any)

kelvin.sdk.lib.utils.general_utils.chdir(path)[source]¶

Change working directory and return to previous on exit.

Return type:: Any
Parameters:: path (Path | None)

kelvin.sdk.lib.utils.general_utils.is_port_open(host, port)[source]¶

Check if a port is being used on a specific ip address

Return type:

bool

Parameters:

host (str)
port (int)

kelvin.sdk.lib.utils.general_utils.get_system_information(pretty_keys=False)[source]¶

Get a dictionary with system-wide information.

Return type:: dict
Parameters:: pretty_keys (bool)

Parameters¶

pretty_keys: bool: indicates whether it should yield a dictionary with raw data or a ‘prettified’ one.

Returns¶

dict:: a dictionary containing all system information.

kelvin.sdk.lib.utils.general_utils.ansi_escape_string(value)[source]¶

When provided with a string, ansi-escape it.

Return type:: str
Parameters:: value (str)

Parameters¶

valuestr: the string to ansi escape

Returns¶

str:: The original, clean string.

kelvin.sdk.lib.utils.general_utils.get_random_hex_string()[source]¶

Yield a random hex string.

Return type:: str

Returns¶

str:: a random hex string

Application Utilities¶

Utility functions for application management and validation.

Licensed under the Kelvin Inc. Developer SDK License Agreement (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.kelvininc.com/developer-sdk-license

kelvin.sdk.lib.utils.application_utils.check_if_app_name_is_valid(app_name)[source]¶

Verify whether the provided app name is valid (or contains a forbidden word combination).

Raise an exception if the provided app name contains a forbidden keyword.

Return type:: bool
Parameters:: app_name (str)

Parameters¶

app_namestr: the app name to be verified.

Returns¶

bool:: a boolean indicating whether the app name is valid.

Click Utilities¶

Utility functions and classes for Click CLI framework customizations.

Licensed under the Kelvin Inc. Developer SDK License Agreement (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.kelvininc.com/developer-sdk-license

class kelvin.sdk.lib.utils.click_utils.ClickConfigs[source]¶

Bases: object

all_verbose_commands = ['--verbose', '-v']¶

all_help_commands = ['--help', '-h']¶

class kelvin.sdk.lib.utils.click_utils.KSDKGroup(name=None, commands=None, invoke_without_command=False, no_args_is_help=None, subcommand_metavar=None, chain=False, result_callback=None, **kwargs)[source]¶

Bases: Group

Parameters:

name (str | None)
commands (cabc.MutableMapping[str, Command] | cabc.Sequence[Command] | None)
invoke_without_command (bool)
no_args_is_help (bool | None)
subcommand_metavar (str | None)
chain (bool)
result_callback (t.Callable[..., t.Any] | None)
kwargs (t.Any)

command_tree: dict = {'agent': {'sync': 'Sync the AGENTS.md file to the application directory.'}, 'app': {'agent': 'Manage AI agent configuration files for applications.', 'build': 'Build a local application into a packaged image.', 'create': 'Create a new application based on the specified parameters.', 'images': 'Management and display of local images on the Local Application Registry.', 'mlflow': 'MLFlow integration commands.', 'samples': "Opens Kelvin's code samples GitHub repo.", 'test': 'Test local applications.', 'upload': "Upload an application to the platform's Application Registry."}, 'appregistry': {'delete': "(Deprecated) Delete an application from the platform's Application Registry.", 'download': '(Deprecated) Download an application from the platform and make it available locally.', 'list': "(Deprecated) List all the available applications on the platform's Application registry.", 'search': "(Deprecated) Search for specific apps on the platform's Application Registry.", 'show': '(Deprecated) Show the platform details and configurations for a specific application.', 'upload': "(Deprecated) Upload an application to the platform's Application Registry."}, 'apps': {'delete': "Delete an application from the platform's Application Registry.", 'download': 'Download an application from the platform and make it available locally.', 'list': "List all the available applications on the platform's Application registry.", 'search': "Search for specific apps on the platform's Application Registry.", 'show': 'Show the platform details and configurations for a specific application.', 'upload': "Upload an application to the platform's Application Registry."}, 'auth': {'login': 'Login on the platform.', 'logout': 'Logout from the platform.', 'reset': 'Reset all authentication and configuration cache.', 'token': 'Obtain an authentication token for the platform.'}, 'configuration': {'autocomplete': 'Generate command-completion configuration for KSDK commands.', 'list': 'List all the available configurations for this tool.', 'set': 'Set a local configuration for this tool.', 'unset': 'Unset a local configuration for this tool.'}, 'images': {'list': 'List all locally built applications.', 'remove': 'Remove an application from the local applications list.', 'unpack': 'Extract the content of an application from its built image into the specified directory.'}, 'ksdk': {'app': 'Create, build and manage applications.', 'appregistry': '(Deprecated) Manage platform Applications.', 'apps': 'Manage platform Applications.', 'auth': 'Platform authentication.', 'configuration': 'Local configurations that enhance the usage of this tool.', 'info': 'Provide system information and the currently logged platform.', 'reset': 'Reset all configurations & cache used by Kelvin SDK.', 'secret': "Manage platform 'secrets'.", 'workload': 'Manage and view application workloads.'}, 'mlflow': {'create': 'Create app based on MLFlow model.', 'import': 'Import a MLFlow model into an existing Kelvin App.', 'list': 'Search MLFlow models.'}, 'secret': {'create': 'Create a secret on the platform.', 'delete': 'Delete secrets on the platform.', 'list': 'List all the available secrets on the platform.', 'update': 'Update an existing secret on the platform.'}, 'test': {'csv': 'Publishes data from a csv to the application.', 'generator': 'Publishes data generated by a custom generator class', 'simulator': "Generates random data to application's inputs"}, 'workload': {'deploy': 'Deploy a workload with specified parameters.', 'list': 'List the workloads available on the platform.', 'logs': 'Display the logs of a specific workload.', 'search': 'Search for specific workloads.', 'show': 'Show the details of a specific workload.', 'start': 'Start a workload on a node.', 'stop': 'Stop a running workload.', 'undeploy': 'Undeploy and delete a workload.', 'update': 'Update a specific workload based with new configurations.'}}¶

add_command(cmd, name=None)[source]¶

Registers another Command with this group. If the name is not provided, the name of the command is used.

Return type:

None

Parameters:

cmd (Any)
name (Any)

add_command_alias(cmd, name)[source]¶

Return type:

None

Parameters:

cmd (Any)
name (str)

get_command_tree()[source]¶

Return type:: dict

class kelvin.sdk.lib.utils.click_utils.KSDKCommand(name, context_settings=None, callback=None, params=None, help=None, epilog=None, short_help=None, options_metavar='[OPTIONS]', add_help_option=True, no_args_is_help=False, hidden=False, deprecated=False)[source]¶

Bases: Command

Parameters:

name (str | None)
context_settings (cabc.MutableMapping[str, t.Any] | None)
callback (t.Callable[..., t.Any] | None)
params (list[Parameter] | None)
help (str | None)
epilog (str | None)
short_help (str | None)
options_metavar (str | None)
add_help_option (bool)
no_args_is_help (bool)
hidden (bool)
deprecated (bool | str)

version_warning_message: str | None = None¶

static get_verbose_option(_)[source]¶

Return type:: Option
Parameters:: _ (Context)

get_params(ctx)[source]¶

Return type:: List
Parameters:: ctx (Context)

parse_args(ctx, args)[source]¶

Return type:

list

Parameters:

ctx (Any)
args (Any)

invoke(ctx)[source]¶

Given a context, this invokes the attached callback (if it exists) in the right way.

Return type:: Any
Parameters:: ctx (Any)

class kelvin.sdk.lib.utils.click_utils.ClickExpandedPath(exists=False, file_okay=True, dir_okay=True, writable=False, readable=True, resolve_path=False, allow_dash=False, path_type=None, executable=False)[source]¶

Bases: Path

A Click path argument that returns a Path, not a string.

Parameters:

exists (bool)
file_okay (bool)
dir_okay (bool)
writable (bool)
readable (bool)
resolve_path (bool)
allow_dash (bool)
path_type (type[t.Any] | None)
executable (bool)

convert(value, param=None, ctx=None)[source]¶

Return a Path from the string click would have created with the given options.

Return type:

Any

Parameters:

value (str)
param (Parameter | None)
ctx (Context | None)

class kelvin.sdk.lib.utils.click_utils.AppNameWithVersionType(version_required=False, allow_container=False)[source]¶

Bases: StringParamType

Click Custom App name with version validator Validates for app name with version “someapp:1.0.0” and/or container name “somecontainer.app”

Attributes¶

version_required: bool, optional, default False: If True, version needs to be present in the string e.g. “app:1.0.0”
allow_container: bool, optional, default False: If True, also the container name is allowed in the following format “somecontainer.app”

convert(value, param, ctx)[source]¶

Convert the value to the correct type. This is not called if the value is None (the missing value).

This must accept string values from the command line, as well as values that are already the correct type. It may also convert other compatible types.

The param and ctx arguments may be None in certain situations, such as when converting prompt input.

If the value cannot be converted, call fail() with a descriptive message.

Parameters:

value (str) – The value to convert.
param (Optional[Parameter]) – The parameter that is using this type to convert its value. May be None.
ctx (Optional[Context]) – The current context that arrived at this value. May be None.

Return type:

str

validate_container(value, param, ctx)[source]¶

Return type:

None

Parameters:

value (str)
param (Parameter | None)
ctx (Context | None)

validate_app_name_with_version(value, param, ctx)[source]¶

Return type:

None

Parameters:

value (str)
param (Parameter | None)
ctx (Context | None)

Parameters:

version_required (bool)
allow_container (bool)

Display Utilities¶

Utility functions for formatting and displaying output to the console.

Licensed under the Kelvin Inc. Developer SDK License Agreement (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.kelvininc.com/developer-sdk-license

class kelvin.sdk.lib.utils.display_utils.DisplayObject(raw_data, tabulated_data)[source]¶

Bases: object

Parameters:

raw_data (Any)
tabulated_data (str)

static parse_entry(entry)[source]¶

Return type:: List
Parameters:: entry (Any)

property parsed_data: List[dict]¶

The DisplayObject is usually associated with a model info. Treat the ‘parsed data’ as the dict representation of that model. No need to process the data if its already in its dict format.

Returns¶

List[dict]: the parsed data in a list of dictionaries.

kelvin.sdk.lib.utils.display_utils.display_yes_or_no_question(question, default='no')[source]¶

Ask a yes/no question via raw_input() and return their answer.

It must be “yes” (the default), “no” or None (meaning an answer is required of the user).

Return type:

bool

Parameters:

question (str)
default (str)

Parameters¶

questionstr: is a string that is presented to the user.
defaultstr: is the presumed answer if the user just hits <Enter>.

Returns¶

bool:: The “answer” return value is True for “yes” or False for “no”.

kelvin.sdk.lib.utils.display_utils.display_custom_yes_or_no_question(question, default='no')[source]¶

Ask a yes/no question via raw_input() and return their answer.

It must be “yes” (the default), “no” or None (meaning an answer is required of the user).

Return type:

bool

Parameters:

question (str)
default (str | None)

Parameters¶

questionstr: is a string that is presented to the user.
defaultstr: is the presumed answer if the user just hits <Enter>.

Returns¶

bool:: The “answer” return value is True for “yes” or False for “no”.

kelvin.sdk.lib.utils.display_utils.display_data_entries(data, attributes, header_names, table_title, should_display=True, no_data_message=None)[source]¶

Display generic resource data on the screen.

Will yield the first 10 results (if any is returned) to prevent an overwhelming console output.

If this is the case, a prompt will ask whether the user wishes to see more outputs.

Return type:

DisplayObject

Parameters:

data (List)
attributes (List[str])
header_names (List[str])
table_title (str)
should_display (bool)
no_data_message (str | None)

Parameters¶

data: List: the data to be printed. Usually, a list of dictionaries.
attributes: List[str]: the data attributes.
header_names: List[str]: the ‘pretty’ keys to be displayed atop the table.
table_title: str: the title to be displayed atop the table.
should_display: bool: indicates whether it should print the result.
no_data_message: str: the custom message to be displayed on the table if no data is passed.

Returns¶

DisplayObject:: a symbolic boolean flag indicating the display of entries was successful.

kelvin.sdk.lib.utils.display_utils.display_data_object(data, object_title=None, should_display=True)[source]¶

Displays the provided data object, be it an app or a node, in a simple, yaml-like format.

Return type:

DisplayObject

Parameters:

data (DataModel | Dict)
object_title (str | None)
should_display (bool)

Parameters¶

data: Union[DataModelBase, Dict]: the data object to be displayed on screen.
object_title: Optional[str]: the title to display above the object.
should_display: bool: indicates whether it should print the result.

Returns¶

DisplayObject:: a symbolic boolean flag indicating the display of the object was successful.

kelvin.sdk.lib.utils.display_utils.pretty_colored_content(content, initial_indent=1, indent=1, level=0, show_arm=False)[source]¶

When provided with a dictionary, return the colorized, ‘prettified’ yaml-like object.

Return type:

str

Parameters:

content (dict)
initial_indent (int)
indent (int)
level (int)
show_arm (bool)

Parameters¶

content: dict: the content to ‘prettify’
initial_indent: int: the initial indent dimension to split on.
indent: int: the initial indent dimension to split on.
level: int: variable to track the level of display.
show_arm: bool: indicate whether the ‘pretty arm’ should be displayed before a key.

Returns¶

str:: a ‘pretty’, indented yaml-like object.

kelvin.sdk.lib.utils.display_utils.success_colored_message(message)[source]¶

Yield a ‘green colored’ message.

Return type:: str
Parameters:: message (str)

Parameters¶

messagestr: the message to yield in the success format.

Returns¶

str

kelvin.sdk.lib.utils.display_utils.warning_colored_message(message)[source]¶

Yield a ‘orange colored’ message.

Return type:: str
Parameters:: message (str)

Parameters¶

messagestr: the message to yield in the warning format.

Returns¶

str

kelvin.sdk.lib.utils.display_utils.error_colored_message(message)[source]¶

Yield a ‘red colored’ message.

Return type:: str
Parameters:: message (str)

Parameters¶

messagestr: the message to yield in the error format.

Returns¶

str

Deploy Bulk Utilities¶

Utility functions for bulk deployment operations.

kelvin.sdk.lib.utils.deploy_bulk_utils.load_workload_data(file, file_type, field_map=None)[source]¶

Return type:

Sequence[Mapping[str, Any]]

Parameters:

file (KPath)
file_type (WorkloadFileType)
field_map (Mapping[str, Any] | None)

kelvin.sdk.lib.utils.deploy_bulk_utils.save_workload_data(file, filename, file_type, output_filename, results)[source]¶

Return type:

bool

Parameters:

file (KPath)
filename (str)
file_type (WorkloadFileType)
output_filename (str | None)
results (Sequence[Mapping])

Exception Utilities¶

Utility functions for exception handling and error formatting.

kelvin.sdk.lib.utils.exception_utils.retrieve_error_message_from_api_exception(api_error)[source]¶

Returns the ‘pretty’ error message from the APIError.

Return type:: str
Parameters:: api_error (APIError)

Parameters¶

api_errorAPIError: The exception yielded by the service call.

Returns¶

str:: a string containing the error message of the APIError.

kelvin.sdk.lib.utils.exception_utils.retrieve_error_message_from_keycloak_exception(keycloak_exception)[source]¶

Returns the ‘pretty’ error message from the KeycloakError.

Return type:: str
Parameters:: keycloak_exception (KeycloakError)

Parameters¶

keycloak_exceptionKeycloakError: The exception yielded by the service call.

Returns¶

str:: a string containing the error message of the KeycloakError.

Logger Utilities¶

Utility functions for logging configuration and management.

Licensed under the Kelvin Inc. Developer SDK License Agreement (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.kelvininc.com/developer-sdk-license

class kelvin.sdk.lib.utils.logger_utils.LoggerProtocol(*args, **kwargs)[source]¶

Bases: Protocol

debug(event=None, *args, **kw)[source]¶

Return type:

Any

Parameters:

event (str | None)
args (Any)
kw (Any)

info(event=None, *args, **kw)[source]¶

Return type:

Any

Parameters:

event (str | None)
args (Any)
kw (Any)

warning(event=None, *args, **kw)[source]¶

Return type:

Any

Parameters:

event (str | None)
args (Any)
kw (Any)

error(event=None, *args, **kw)[source]¶

Return type:

Any

Parameters:

event (str | None)
args (Any)
kw (Any)

relevant(event=None, *args, **kw)[source]¶

Return type:

Any

Parameters:

event (str | None)
args (Any)
kw (Any)

exception(event=None, *args, **kw)[source]¶

Return type:

Any

Parameters:

event (str | None)
args (Any)
kw (Any)

class kelvin.sdk.lib.utils.logger_utils.BaseLogFormatter(fmt=None, datefmt=None, style='%', validate=True, *, defaults=None)[source]¶: Bases: Formatter

class kelvin.sdk.lib.utils.logger_utils.KSDKLogFormatter(log_color=None, debug=False)[source]¶

Bases: BaseLogFormatter

Logging Formatter to add colors and count warning / errors

Parameters:

log_color (LogColor | None)
debug (bool)

message_format = '[%(logger)s][%(asctime)s][%(levelname).1s] %(message)s'¶

FORMATS = {LogColor.COLORED: {10: '\x1b[36m[%(logger)s][%(asctime)s][%(levelname).1s] %(message)s\x1b[39m', 20: '\x1b[39m[%(logger)s][%(asctime)s][%(levelname).1s] %(message)s\x1b[39m', 25: '\x1b[32m[%(logger)s][%(asctime)s][%(levelname).1s] %(message)s\x1b[39m', 30: '\x1b[33m[%(logger)s][%(asctime)s][%(levelname).1s] %(message)s\x1b[39m', 40: '\x1b[31m[%(logger)s][%(asctime)s][%(levelname).1s] %(message)s\x1b[39m', 50: '\x1b[31m[%(logger)s][%(asctime)s][%(levelname).1s] %(message)s\x1b[39m'}, LogColor.COLORLESS: {10: '[%(logger)s][%(asctime)s][%(levelname).1s] %(message)s', 20: '[%(logger)s][%(asctime)s][%(levelname).1s] %(message)s', 25: '[%(logger)s][%(asctime)s][%(levelname).1s] %(message)s', 30: '[%(logger)s][%(asctime)s][%(levelname).1s] %(message)s', 40: '[%(logger)s][%(asctime)s][%(levelname).1s] %(message)s', 50: '[%(logger)s][%(asctime)s][%(levelname).1s] %(message)s'}}¶

format(record)[source]¶

Format the specified record as text.

The record’s attribute dictionary is used as the operand to a string formatting operation which yields the returned string. Before formatting the dictionary, a couple of preparatory steps are carried out. The message attribute of the record is computed using LogRecord.getMessage(). If the formatting string uses the time (as determined by a call to usesTime(), formatTime() is called to format the event time. If there is exception information, it is formatted using formatException() and appended to the message.

Return type:: str
Parameters:: record (Any)

class kelvin.sdk.lib.utils.logger_utils.JSONLogFormatter(debug=False)[source]¶

Bases: BaseLogFormatter

Parameters:: debug (bool)

format(record)[source]¶

Format the specified record as text.

Return type:: str
Parameters:: record (Any)

kelvin.sdk.lib.utils.logger_utils.relevant(parent, msg, *args, **kwargs)[source]¶

Return type:

None

Parameters:

parent (Any)
msg (str)
args (Any)
kwargs (Any)

kelvin.sdk.lib.utils.logger_utils.setup_logger(log_type=LogType.KSDK, log_color=LogColor.COLORED, verbose=False, debug=False, history_file=None)[source]¶

Return type:

LoggerProtocol

Parameters:

log_type (LogType)
log_color (LogColor)
verbose (bool)
debug (bool)
history_file (Path | None)

Version Utilities¶

Utility functions for version parsing and comparison.

Licensed under the Kelvin Inc. Developer SDK License Agreement (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.kelvininc.com/developer-sdk-license

kelvin.sdk.lib.utils.version_utils.assess_version_status(current_version, minimum_version, latest_version, should_warn=True)[source]¶

Verifies whether the current KSDK version is supported.

Warn the user, if ‘should_warn’ is set to True, that the SDK is outdated. Raise an exception should it not respect the minimum version.

Return type:

VersionStatus

Parameters:

current_version (str)
minimum_version (str)
latest_version (str)
should_warn (bool)

Parameters¶

current_versionstr: the current version to check.
minimum_versionstr: the minimum accepted version.
latest_versionstr: the latest version of the SDK.
should_warnbool: if set to true, will warn the user in case the ksdk is out of version.

Returns¶

VersionStatus:: the corresponding version status.

kelvin.sdk.lib.utils.version_utils.check_if_is_pre_release(v)[source]¶

Check if a given version is a pre-release

Return type:: bool
Parameters:: v (str)

Parameters¶

versionstr: the version to verify.

Returns¶

bool:: a bool indicating whether the version matches the indicated version type.

PyPI Utilities¶

Utility functions for PyPI package management operations.

Licensed under the Kelvin Inc. Developer SDK License Agreement (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.kelvininc.com/developer-sdk-license

kelvin.sdk.lib.utils.pypi_utils.get_pypi_credentials()[source]¶

When provided with Kelvin SDK global configuration, retrieve the pypi credentials to install external dependencies required by the application.

Return type:: dict

Returns¶

dict:: a dictionary containing the required pypi credentials for external dependencies.

kelvin.sdk.lib.utils.pypi_utils.get_pretty_source_from_section(section, pypi_url)[source]¶

Return the pretty log entry for a given section & pypi url.

Return type:

str

Parameters:

section (str)
pypi_url (str)

Parameters¶

section :str: the section (env/pip) corresponding to the pip credential source.
pypi_url :str: the pypi url associated to the section.

Returns¶

str:: the redacted version of the original url.

kelvin.sdk.lib.utils.pypi_utils.redact_pip_url(pip_url)[source]¶

Hide sensitive data (password) from a valid PyPi credential.

Return type:: str
Parameters:: pip_url (str)

Parameters¶

pip_url :str: the pip url to redact to obfuscate.

Returns¶

str:: the original, redacted pip url