tuneapi.utils package

Submodules

tuneapi.utils.code module

class tuneapi.utils.code.Var(name: str = '', description: str = '', required: bool = False, type: str | List[Var] = '', additionalProperties: List[Var] | Var = [])

Bases: object

classmethod from_dict(d: Dict[str, Any]) → Var

Deserialise a Var from a dictionary.

Parameters:: d (Dict[str, Any]) – The dictionary to deserialise from.
Returns:: The deserialised Var.
Return type:: Var

to_dict() → Dict[str, Any]

Serialise this Var to a dictionary that can be JSON serialised and sent to the client.

Returns:: The serialised Var.
Return type:: Dict[str, Any]

tuneapi.utils.code.func_to_vars(func: object, log_trace: bool = False) → List[Var]

Extracts the signature of a function and converts it to an array of Var objects.

Parameters:: func (Callable) – The function to extract the signature from.
Returns:: The array of Var objects.
Return type:: List[Var]

tuneapi.utils.code.pyannotation_to_json_schema(x: Any, allow_any: bool, allow_exc: bool, allow_none: bool, *, trace: bool = False, is_return: bool = False) → Var

Function to convert the given annotation from python to a Var which can then be JSON serialised and sent to the clients.

Parameters:

x (Any) – The annotation to convert.
allow_any (bool) – Whether to allow the Any type.
allow_exc (bool) – Whether to allow the Exception type.
allow_none (bool) – Whether to allow the None type.
trace (bool, optional) – Adds verbosity the schema generation. Defaults to False.

Returns:

The converted annotation.

Return type:

Var

tuneapi.utils.env module

tuneapi.utils.env.ENV = <tuneapi.utils.env._ENV object>

Convinience class to get environment variables

Usage:

from tuneapi.utils.env import ENV
TUNEAPI_TOKEN = ENV.TUNEAPI_TOKEN()   # get defined values
MY_ENV = ENV.MY_ENV()               # get your arbitrary env var

tuneapi.utils.fs module

tuneapi.utils.fs.fetch(url, cache='/tmp', method='post', force: bool = False, **kwargs): fetch a url and cache it

tuneapi.utils.fs.folder(x: str) → str: get the folder of this file path

tuneapi.utils.fs.get_files_in_folder(folder, ext='*', recursive: bool = False, ig_pat: str = '', abs_path: bool = True, followlinks: bool = False) → List[str]: Get files with ext in folder

tuneapi.utils.fs.joinp(x: str, *args) → str: convienience function for os.path.join

tuneapi.utils.fs.list_dir(folder, ext='*', recursive: bool = False, ig_pat: str = '', abs_path: bool = True, followlinks: bool = False) → List[str]: Alias for get_files_in_folder

tuneapi.utils.fs.load_module_from_path(fn_name, file_path)

tuneapi.utils.logger module

tuneapi.utils.logger.get_logger(name: str = 'tuneapi') → Logger: Returns a logger object

tuneapi.utils.logger.logger = <Logger tuneapi (INFO)>: Logger provided with the package, try to use this logger for all logging purposes

tuneapi.utils.logger.warning_with_fix(msg: str, fix: str | None)

tuneapi.utils.logic module

JSON Logic used from https://github.com/nadirizr/json-logic-py (MIT License)

tuneapi.utils.logic.get_var(data, var_name, not_found=None): Gets variable value from data dictionary.

tuneapi.utils.logic.hard_equals(a, b): Implements the ‘===’ operator.

tuneapi.utils.logic.if_(*args): Implements the ‘if’ operator with support for multiple elseif-s.

tuneapi.utils.logic.json_logic(tests, data=None): Executes the json-logic with given data.

tuneapi.utils.logic.less(a, b, *args): Implements the ‘<’ operator with JS-style type coertion.

tuneapi.utils.logic.less_or_equal(a, b, *args): Implements the ‘<=’ operator with JS-style type coertion.

tuneapi.utils.logic.merge(*args): Implements the ‘merge’ operator for merging lists.

tuneapi.utils.logic.minus(*args): Also, converts either to ints or to floats.

tuneapi.utils.logic.missing(data, *args): Implements the missing operator for finding missing variables.

tuneapi.utils.logic.missing_some(data, min_required, args): Implements the missing_some operator for finding missing variables.

tuneapi.utils.logic.plus(*args): Sum converts either to ints or to floats.

tuneapi.utils.logic.soft_equals(a, b): Implements the ‘==’ operator, which does type JS-style coertion.

tuneapi.utils.logic.to_numeric(arg): Converts a string either to int or to float. This is important, because e.g. {“!==”: [{“+”: “0”}, 0.0]}

tuneapi.utils.mime module

tuneapi.utils.mime.get_mime_type(fp: str, defualt='application/octet-stream'): Get mime type from file path based on extension

tuneapi.utils.misc module

class tuneapi.utils.misc.SimplerTimes

Bases: object

A class that provides a simpler interface to datetime and time modules.

IST = datetime.timezone(datetime.timedelta(seconds=19800))

get_now_datetime() → datetime: Get the current datetime in UTC timezone

get_now_float() → float: Get the current datetime in UTC timezone as a float

get_now_fp64() → float: Get the current datetime in UTC timezone as a float

get_now_human(tz=None) → str: Get the current datetime in timezone as a human readable string

get_now_i64() → int: Get the current datetime in UTC timezone as a int

get_now_ns() → int

get_now_pb()

get_now_str() → str: Get the current datetime in UTC timezone as a string

i64_to_datetime() → datetime: Convert an int to datetime in UTC timezone

tz = datetime.timezone.utc

tuneapi.utils.misc.decrypt(token: str, password: str, salt: str)

tuneapi.utils.misc.encrypt(text: str, password: str, salt: str)

tuneapi.utils.misc.generator_to_api_events(model, generator): Takes in a token generator and creates OpenAI compatible event stream

tuneapi.utils.misc.hashstr(item: str, fn='md5'): Hash sting of any item

tuneapi.utils.misc.safe_exit(code=0): This function is used to exit the program safely. It is used to make sure that the program exits properly and all the resources are cleaned up.

tuneapi.utils.misc.unsafe_exit(code=0): why use os._exit over sys.exit: https://stackoverflow.com/questions/9591350/what-is-difference-between-sys-exit0-and-os-exit0 https://stackoverflow.com/questions/19747371/python-exit-commands-why-so-many-and-when-should-each-be-used tl;dr: os._exit kills without cleanup and so it’s okay on the Pod

tuneapi.utils.networking module

exception tuneapi.utils.networking.DoNotRetryException

Bases: Exception

Raised when code tells not to retry

exception tuneapi.utils.networking.UnAuthException

Bases: Exception

Raised when the API returns a 401

tuneapi.utils.networking.exponential_backoff(foo, *args, max_retries=2, retry_delay=1, **kwargs) → Dict[str, Any]

Exponential backoff function

Parameters:

foo (function) – The function to call
max_retries (int, optional) – maximum number of retries. Defaults to 2.
retry_delay (int, optional) – Initial delay in seconds. Defaults to 1.

Raises:

e – Max retries reached. Exiting…
Exception – This should never happen

Returns:

The completion(s) generated by the API.

Return type:

Dict[str, Any]

tuneapi.utils.parallel module

tuneapi.utils.parallel.batched(iterable, n, ol=0, expand: bool = False, last: bool = True)

Convert any iterable to a generator of batches of size n, last one may be smaller. Python 3.12 has itertools.batched which does the same thing.

Example

>>> for x in batched(range(10), 3):
...    print(x)
[0, 1, 2]
[3, 4, 5]
[6, 7, 8]
[9]

>>> for x in batched(range(10), 3, ol = 1): # overlap = 1
...    print(x)
[0, 1, 2]
[2, 3, 4]
[4, 5, 6]
[6, 7, 8]
[8, 9]

>>> for x in batched(range(10), 3, last = False):
...    print(x)
[0, 1, 2]
[3, 4, 5]
[6, 7, 8]

Parameters:

iterable (Iterable) – The iterable to convert to batches
n (int) – The batch size
ol (int) – amount of overlap between batches
expand (bool, optional) – If true, each item in batch is tuple, eg. in numpy x[ ... , None]
last (bool, optional) – If true, return the last batch even if it is smaller than n. Defaults to True.

Yields:

Iterator – The batched iterator

tuneapi.utils.parallel.threaded_map(fn, inputs: List[Tuple] | Generator, wait: bool = True, max_threads=20, post_fn=None, _name: str = '', safe: bool = False, pbar: bool = False) → Dict[Future, int] | List[Any]

inputs is a list of tuples, each tuple is the input for single invocation of fn. order is preserved.

Parameters:

fn (function) – The function to call
inputs (List[Tuple[Any]]) – All the inputs to the function, can be a generator
wait (bool, optional) – If true, wait for all the threads to finish, otherwise return a dict of futures. Defaults to True.
max_threads (int, optional) – The maximum number of threads to use. Defaults to 20.
post_fn (function, optional) – A function to call with the result. Defaults to None.
_name (str, optional) – The name of the thread pool. Defaults to “”.
safe (bool, optional) – If true, all exceptions are caught and returned with results. Defaults to False.
pbar (bool, optional) – If true, show a progress bar. Defaults to False.

tuneapi.utils.randomness module

class tuneapi.utils.randomness.SFGen(instance: int = None, epoch=1705905900000)

Bases: object

CURRENT_EPOCH_START = 1705905900000: Start of the current epoch, used for generating snowflake ids

tuneapi.utils.randomness.get_random_string(length: int, numbers: bool = True, special: bool = False) → str

tuneapi.utils.randomness.reservoir_sampling(stream, k)

Perform reservoir sampling on the given stream to select k items.

Parameters:

stream – An iterable representing the input stream.
k – The number of items to sample.

Returns:

A list containing k sampled items.

tuneapi.utils.serdeser module

tuneapi.utils.serdeser.dict_to_structpb(data: Dict)

tuneapi.utils.serdeser.from_b64(x: str)

tuneapi.utils.serdeser.from_json(x: str = '') → Dict[str, Any]

Load a JSON string or filepath and return a dictionary.

Parameters:: fp (str) – The filepath or JSON-ified string

Returns:

tuneapi.utils.serdeser.from_pickle(path)

Load an object from a pickle file

Parameters:: path – path to load from

tuneapi.utils.serdeser.from_s3(key: str, aws_region: str, aws_access_key_id: str, aws_secret_access_key: str, aws_s3_bucket: str)

tuneapi.utils.serdeser.structpb_to_dict(struct, out: Dict = None) → Dict

tuneapi.utils.serdeser.to_b64(x: bytes)

tuneapi.utils.serdeser.to_json(x: dict, fp: str = '', indent=2, tight: bool = False) → str

Convert a dict to json string and write to file if fp is provided.

Parameters:

x (dict) – The dict to convert
fp (str, optional) – The file path to write to. Defaults to “”.
indent (int, optional) – The indentation level. Defaults to 2.
tight (bool, optional) – If true, remove all the whitespaces, ignores indent. Defaults to False.

Returns:

The json string if fp is not provided

Return type:

Optional[str]

tuneapi.utils.serdeser.to_pickle(obj, path)

Save an object to a pickle file

Parameters:

obj – object to save
path – path to save to

tuneapi.utils.serdeser.to_s3(key: str, x: Any, aws_region: str, aws_access_key_id: str, aws_secret_access_key: str, aws_s3_bucket: str, content_type: str = None)

tuneapi.utils.subway module

class tuneapi.utils.subway.Subway(_url: str, _session: Session)

Bases: object

Simple code that allows writing APIs by .attr.ing them. This is inspired from gRPC style functional calls which hides the complexity of underlying networking. This is useful when you are trying to debug live server directly.

Note

User is solely responsible for checking if the certain API endpoint exists or not. This simply wraps the API calls and does not do any validation.

Parameters:

_url (str) – The url to use for the client
_session (requests.Session) – The session to use for the client

u(attr: str) → Subway

In cases where the api might start with a number you cannot write in python, this method can be used to access the attribute.

Example

>>> stub.9jisjfi      # python will cry, invalid syntax: cannot start with a number
>>> stub.u('9jisjfi') # do this instead

Parameters:: attr (str) – The attribute to access
Returns:: The new subway object
Return type:: Subway

exception tuneapi.utils.subway.SubwayClientError(*args, code: str)

Bases: Exception

Raised if 399 < status_code < 500

exception tuneapi.utils.subway.SubwayServerError(*args, code: str)

Bases: Exception

Raised if 499 < status_code < 600

tuneapi.utils.subway.get_session(token: str | None = '', bearer: str | None = '', **headers) → Session

tuneapi.utils.subway.get_subway(url='http://localhost:8080', headers: Dict[str, Any] | None = None) → Subway

tuneapi.utils.terminal module

class tuneapi.utils.terminal.color

Bases: object

black(b=False)

blue(b=False)

bold()

cyan(b=False)

green(b=False)

magenta(b=False)

red(b=False)

white(b=False)

yellow(b=False)

tuneapi.utils.terminal.hr(msg: str = '') → str

Prints full wodth text message on the terminal

Parameters:: msg (str, optional) – The message to print. Defaults to “”.
Returns:: The message to print
Return type:: str

tuneapi.utils package

Submodules

tuneapi.utils.code module

tuneapi.utils.env module

tuneapi.utils.fs module

tuneapi.utils.logger module

tuneapi.utils.logic module

tuneapi.utils.mime module

tuneapi.utils.misc module

tuneapi.utils.networking module

tuneapi.utils.parallel module

tuneapi.utils.randomness module

tuneapi.utils.serdeser module

tuneapi.utils.subway module

tuneapi.utils.terminal module

Module contents