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 is already exist, 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.

call run_async(my_function, any_event_loop=True, *args, **kwargs) to enable run with any eventloop

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.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

jina.helper.docarray_graphql_compatible()[source]#

Check if installed docarray version is compatible with GraphQL features.

Returns: True if compatible, False if not