jina.helper module#

jina.helper.batch_iterator(data, batch_size, axis=0)[source]#

Get an iterator of batches of data.

For example: .. highlight:: python .. code-block:: python

for req in batch_iterator(data, batch_size, split_over_axis):
pass # Do something with batch

Parameters:

data (Iterable[Any]) – Data source.
batch_size (int) – Size of one batch.
axis (int) – Determine which axis to iterate for np.ndarray data.

Yield:

data

Return type:

Iterator[Any]

Returns:

An Iterator of batch data.

jina.helper.parse_arg(v)[source]#

Parse the arguments from string to Union[bool, int, str, list, float].

Parameters:: v (str) – The string of arguments
Return type:: Union[bool, int, str, list, float, None]
Returns:: The parsed arguments list.

jina.helper.random_port()[source]#

Get a random available port number.

Return type:: Optional[int]
Returns:: A random port.

jina.helper.random_identity(use_uuid1=False)[source]#

Generate random UUID.

..note::

A MAC address or time-based ordering (UUID1) can afford increased database performance, since it’s less work to sort numbers closer-together than those distributed randomly (UUID4) (see here).

A second related issue, is that using UUID1 can be useful in debugging, even if origin data is lost or not explicitly stored.

Parameters:: use_uuid1 (bool) – use UUID1 instead of UUID4. This is the default Document ID generator.
Return type:: str
Returns:: A random UUID.

jina.helper.random_uuid(use_uuid1=False)[source]#

Get a random UUID.

Parameters:: use_uuid1 (bool) – Use UUID1 if True, else use UUID4.
Return type:: UUID
Returns:: A random UUID.

jina.helper.expand_env_var(v)[source]#

Expand the environment variables.

Parameters:: v (str) – String of environment variables.
Return type:: Union[bool, int, str, list, float, None]
Returns:: Parsed environment variables.

jina.helper.colored(text, color=None, on_color=None, attrs=None)[source]#

Give the text with color.

Parameters:

text (str) – The target text.
color (Optional[str]) –
The color of text. Chosen from the following. {

’grey’: 30, ‘red’: 31, ‘green’: 32, ‘yellow’: 33, ‘blue’: 34, ‘magenta’: 35, ‘cyan’: 36, ‘white’: 37

}
on_color (Optional[str]) –
The on_color of text. Chosen from the following. {

’on_grey’: 40, ‘on_red’: 41, ‘on_green’: 42, ‘on_yellow’: 43, ‘on_blue’: 44, ‘on_magenta’: 45, ‘on_cyan’: 46, ‘on_white’: 47

}
attrs (Union[str, list, None]) –
Attributes of color. Chosen from the following. {

’bold’: 1, ‘dark’: 2, ‘underline’: 4, ‘blink’: 5, ‘reverse’: 7, ‘concealed’: 8

}

Return type:

str

Returns:

Colored text.

class jina.helper.ArgNamespace[source]#

Bases: object

Helper function for argparse.Namespace object.

static kwargs2list(kwargs)[source]#

Convert dict to an argparse-friendly list.

Parameters:: kwargs (Dict) – dictionary of key-values to be converted
Return type:: List[str]
Returns:: argument list

static kwargs2namespace(kwargs, parser, warn_unknown=False, fallback_parsers=None, positional_args=None)[source]#

Convert dict to a namespace.

Parameters:

kwargs (Dict[str, Union[str, int, bool]]) – dictionary of key-values to be converted
parser (ArgumentParser) – the parser for building kwargs into a namespace
warn_unknown (bool) – True, if unknown arguments should be logged
fallback_parsers (Optional[List[ArgumentParser]]) – a list of parsers to help resolving the args
positional_args (Optional[Tuple[str, ...]]) – some parser requires positional arguments to be presented

Return type:

Namespace

Returns:

argument list

static get_non_defaults_args(args, parser, taboo=None)[source]#

Get non-default args in a dict.

Parameters:

args (Namespace) – the namespace to parse
parser (ArgumentParser) – the parser for referring the default values
taboo (Optional[Set[str]]) – exclude keys in the final result

Return type:

Dict

Returns:

non defaults

static flatten_to_dict(args)[source]#

Convert argparse.Namespace to dict to be uploaded via REST.

Parameters:: args (Union[Dict[str, Namespace], Namespace]) – namespace or dict or namespace to dict.
Return type:: Dict[str, Any]
Returns:: pod args

jina.helper.is_valid_local_config_source(path)[source]#

Check if the path is valid.

Parameters:: path (str) – Local file path.
Return type:: bool
Returns:: True if the path is valid else False.

class jina.helper.cached_property(func)[source]#

Bases: object

The decorator to cache property of a class.

Create the cached_property.

Parameters:: func – Cached function.

jina.helper.typename(obj)[source]#

Get the typename of object.

Parameters:: obj – Target object.
Returns:: Typename of the obj.

jina.helper.get_public_ip(timeout=0.3)[source]#

Return the public IP address of the gateway for connecting from other machine in the public network.

Parameters:: timeout (float) – the seconds to wait until return None.
Returns:: Public IP address.

jina.helper.get_internal_ip()[source]#

Return the private IP address of the gateway for connecting from other machine in the same network.

Returns:: Private IP address.

jina.helper.convert_tuple_to_list(d)[source]#

Convert all the tuple type values from a dict to list.

Parameters:: d (Dict) – Dict type of data.

jina.helper.run_async(func, *args, **kwargs)[source]#

Generalized asyncio.run for jupyter notebook.

When running inside jupyter, an eventloop already exists, can’t be stopped, can’t be killed. Directly calling asyncio.run will fail, as This function cannot be called when another asyncio event loop is running in the same thread.

Parameters:

func – function to run
args – parameters
kwargs – key-value parameters

Returns:

asyncio.run(func)

jina.helper.deprecated_alias(**aliases)[source]#

Usage, kwargs with key as the deprecated arg name and value be a tuple, (new_name, deprecate_level).

With level 0 means warning, level 1 means exception.

For example:

@deprecated_alias(
    input_fn=('inputs', 0),
    buffer=('input_fn', 0),
    callback=('on_done', 1),
    output_fn=('on_done', 1),
)
def some_function(inputs, input_fn, on_done):
    pass

Parameters:: aliases – maps aliases to new arguments
Returns:: wrapper

jina.helper.retry(num_retry=3, message='Calling {func_name} failed, retry attempt {attempt}/{num_retry}. Error: {error!r}')[source]#

Retry calling a function again in case of an error.

Parameters:

num_retry (int) – number of times to retry
message (str) – message to log when error happened

Returns:

wrapper

jina.helper.countdown(t, reason='I am blocking this thread')[source]#

Display the countdown in console.

For example:

Parameters:

t (int) – Countdown time.
reason (str) – A string message of reason for this Countdown.

Return type:

None

class jina.helper.CatchAllCleanupContextManager(sub_context)[source]#

Bases: object

This context manager guarantees, that the :method:__exit__ of the sub context is called, even when there is an Exception in the :method:__enter__.

Parameters:: sub_context – The context, that should be taken care of.

jina.helper.download_mermaid_url(mermaid_url, output)[source]#

Download the jpg image from mermaid_url.

Parameters:

mermaid_url – The URL of the image.
output – A filename specifying the name of the image to be created, the suffix svg/jpg determines the file type of the output image.

Return type:

None

jina.helper.get_readable_size(num_bytes)[source]#

Transform the bytes into readable value with different units (e.g. 1 KB, 20 MB, 30.1 GB).

Parameters:: num_bytes (Union[int, float]) – Number of bytes.
Return type:: str
Returns:: Human readable string representation.

jina.helper.get_or_reuse_loop()[source]#

Get a new eventloop or reuse the current opened eventloop.

Returns:: A new eventloop or reuse the current opened eventloop.

jina.helper.get_rich_console()[source]#: Function to get jina rich default console. :return: rich console