utils

This module includes convenient functions

classes

deepdataspace.utils.classes

class SingletonMeta(name, bases, attrs)

Bases: type

Any class whose metaclass is SingletonMeta will be a singleton class.

test()

file

deepdataspace.utils.file

Convenient functions about file system.

file_lock_context(lock_file: str) → bool

Acquire a lock identified by file path. If the file path already exists, return False. If the file path does not exist, create it and return True, and ensure the file is removed on lock releasing.

Parameters:

lock_file – A file path represents the lock.

Example

with file_lock_context(“/tmp/abc.lock”) as ok:

if ok:

do_some_thing()

else:

print(“failed to acquire a lock”)

tailf(file_path: str)

Run a cross-platform ‘tail -f’ command

switch_dir_to_and_back(target_dir: str)

Change to target directory, execute code blocks, and go back to current directory even there is exception in code blocks.

create_file_url(file_path: str, file_encoding: Literal['1', '2'] = ContentEncoding.Plain, read_mode: Literal['t', 'b'] = FileReadMode.Text, file_mime: Optional[str] = None, url_prefix=None) → str

Given the required file info, return a file url to serve the whole file from local disk.

Parameters:

• file_path – the path of the file.

• file_encoding – the encoding of the file, “1” = “plain” or “2” = “base64”. for “plain” encoded file, contents will be read as is. for “base64” encoded file, contents will be read as base64 encoded string, and will be decoded accordingly.

• read_mode – the read mode of the file, “t” or “b”.

• file_mime – the file mime type, e.g. “image/png”. This is used to set the response content type.

• url_prefix – the file url path prefix. If not provided, will use “/files/local_files/{file_id}/{file_name}”. If provided, will use the format: f”{url_prefix}/{file_id}/{file_name}”.

create_file_range_url(file_path: str, beg_pos: int, end_pos: int, file_encoding: Literal['1', '2'] = ContentEncoding.Plain, read_mode: Literal['t', 'b'] = FileReadMode.Text, file_mime: Optional[str] = None, url_prefix: Optional[str] = None) → str

Serve the file from local disk with range. This is similar to serve_whole_file, but it will serve the file with range.

Parameters:

• file_path – the path of the file.

• file_encoding – the encoding of the file, “1” = “plain” or “2” = “base64”. for “plain” encoded file, contents will be read as is. for “base64” encoded file, contents will be read as base64 encoded string, and will be decoded accordingly.

• read_mode – the read mode of the file, “t” or “b”.

• file_mime – the file mime type, e.g. “image/png”. This is used to set the response content type.

• beg_pos – the beginning position of the file.

• end_pos – the end position of the file.

• url_prefix – the file url path prefix. If not provided, will use “/files/local_files/{file_id}/{file_name}”. If provided, will use the format: f”{url_prefix}/{file_id}/{file_name}”.

function

deepdataspace.utils.function

Convenient functions about python function.

count_block_time(block_id: str, logger=print)

Count the time cost of a code block, and log it by the logger. :param block_id: an alias name to the code block. :param logger: the time cost logger, must be a callable, such as logger.info, default is print.

Example

import time

with count_block_time(“test”, print):

time.sleep(1)

>>time cost of block[test]: 1000ms

retry(times: int, sleep: int = 0, exceptions: tuple = (Exception,))

Retry a function or a method. :param times: retry times. :param sleep: sleep time between retries. :param exceptions: the exceptions that will be caught and retry.

Example

@retry(3, 1) def test():

print(“test”) raise Exception

test()

http

deepdataspace.utils.http

Convenient functions and classes for http protocol.

format_response(data: dict, status: int = 200, code: int = 0, msg: str = 'success', enable_cache: bool = False) → Response

Generate a formatted json response with given data. :param data: the data to be returned. :param status: http status code for this response, usually 200 for this function. :param code: err code, usually 0 for success. :param msg: err message, usually “success” for success. :param enable_cache: can this response be cached by browser. :return: A formatted JsonResponse object

exception APIException(code: int, msg: str, http_status: int)

Bases: Exception

to_json_rsp() → Response

Generate a Json Response.

Returns:

Response

raise_exception(code: int, msg: str, status: Optional[int] = None)

Use this function to raise APIException anywhere in the code and return a json response to client directly.

Parameters:

• code – the err code in json response

• msg – the error message in json response

• status – the http status_code for json response, default to err code.

Returns:

None

handle_api_exception(exc: Exception, context) → Response

Catch APIException and return a json response.

Parameters:

• exc – The exception raised.

• context – The exception context.

Returns:

A formatted json response.

class Argument(name: str, type_: callable, location: str, required: bool = False, default=None)

Bases: object

An argument for a request. This helps parse incoming request data. If parse fails, this will raise an APIException directly.

JSON = 'json'

QUERY = 'query'

class Choice(choices: Union[list, set], converter=None)

Bases: object

This represents a choice argument. The value of this argument must be one of the specified choices.

PositiveInt = <deepdataspace.utils.http.Argument._PositiveInt object>

NonNegativeInt = <deepdataspace.utils.http.Argument._NonNegativeInt object>

parse_arguments(request, arguments: List[Argument])

A help function to parse arguments from request data.

class TokenAuthentication

Bases: BaseAuthentication

An authentication class for drf based on UserToken.

authenticate(request)

Authenticate the request and return a two-tuple of (user, token).

class BaseAPIView(**kwargs)

Bases: APIView

Base class for all api views.

class AuthenticatedAPIView(**kwargs)

Bases: BaseAPIView

Base class for all authenticated api views.

authentication_classes = [<class 'deepdataspace.utils.http.TokenAuthentication'>]

import_utils

deepdataspace.utils.import_utils

This file provides convenient functions about importing python packages.

lazy_import(module_name: str, import_hook: Optional[Callable] = None)

class LazyModule(module_name: str, import_hook: Optional[Callable] = None)

Bases: module

import_from_path(path: str)

Import a module given its file path.

network

deepdataspace.utils.network

Convenient functions about network.

get_output_ip_address() → str

Acquire the default output ip address of this machine.

download_by_requests(url: str, path: str, timeout: int = 5)

Download the specified url to the local path.

Parameters:

• url – the remote url to be downloaded

• path – the local path of download destination

• timeout – the timeout for download, in case of endless blocking

This function downloads the url to a temporary path during the download procedure and move the temporary file to the destination path. This guarantees the integrity of the downloaded file.

os

deepdataspace.utils.os

Convenient functions about os.

class Platforms

Bases: object

Mac = 'mac'

MacArm = 'mac_arm'

Win = 'win'

Linux = 'linux'

static is_mac()

A shortcut to test if current platform is Mac

static is_macarm()

A shortcut to test if current platform is Mac Arm

static is_win()

A shortcut to test if current platform is Windows

static is_linux()

A shortcut to test if current platform is Linux

get_platform() → Union[None, str]

Acquire current operation system platform.

get_pid_by_pidfile(pidfile: str) → int

Read the pidfile and check the process status by pid in the file.

Parameters:

pidfile – the pidfile path, which expected to have the process pid inside.

Returns:

int, 0 if process is dead, or pid if process is alive.

get_os_username()

Get the username of current os, works for both windows and linux.

get_pid_by_cmd_id(cmd_id: str, all_user: bool = False) → List[int]

Found all processes whose command line arguments contain the cmd_id, and return their pid in a list.

Parameters:

• cmd_id – A command line string identifier, usually the command line args to start a process.

• all_user – Search for processes of all user. Default value is False, which means only processes of current user will be found.

Returns:

A list of matched process pid.

check_port_free(port: int) → bool

Check if the target port is free for use.

find_free_port(beg_port: int, end_port: int) → Optional[int]

Find a free port in specified port range.

Parameters:

• beg_port – the begen of port range, include itself.

• end_port – the end of port range, include itself.

Returns:

the first free port in range, or None if no port is free.

get_ubuntu_version()

Get the ubuntu release version number. :return: 1804, 2004, 2204, or “” if failed.

find_shared_lib_on_ubuntu(lib_name: str)

find the target shared lib file by name, for ubuntu system only. :params lib_name: the shared lib file name, libssl.so.1.1, libcrypto.so.1.1 etc. :return: lib path, /lib/x86_64-linux-gnu/libssl.so.1.1

find_shared_dirs_on_ubuntu()

find all directories where system search for dynamic libs, for ubuntu system only.

string

deepdataspace.utils.string

Convenient functions about string operation.

get_str_md5(string_: str) → str

Get the md5 hex of target string.

gen_random_str(length: int) → str

Generate a random string of target length. Characters are chosen from ascii_uppercase and digits.

gen_password(length: int) → str

Generate a random string for password. Characters are chosen from ascii_uppercase, digits and punctuations.

encrypt(message: str, key: str) → str

Encrypt message with key, return an encrypted string token.

decrypt(token: str, key: str) → str

Decrypt an encrypted string token with key, return a plain text.