cgnal.core.utils package

Submodules

cgnal.core.utils.cloud module

Module for providing interface to a cloud storage layer that persists files to be used in the application and downloaded on demand.

class cgnal.core.utils.cloud.CloudSync(url: str, root: Union[str, os.PathLike[str]])

Bases: cgnal.core.logging.defaults.WithLogging

Class for sync-ing to a remote file system.

Download a file from the given url and saves it under the given root path.

Parameters

  • url – url to download file from

  • root – local root path under which the downloaded file will be saved

static create_base_directory(filename: Union[str, os.PathLike[str]]) → Union[str, os.PathLike[str]]

Create directory where the given file will be saved.

Parameters

filename – file name, str

Returns

path

get(filename: Union[str, os.PathLike[str]]) → None

Return file from the url.

Parameters

filename – name of file to be downloaded

Raises

FileNotFoundError – if file is not found

get_if_not_exists(filename: Union[str, os.PathLike[str]]) → Union[str, os.PathLike[str]]

Get the file from the remote file system if not exists.

Parameters

filename – relative path

Returns

absolute path

get_if_not_exists_decorator(f: Callable[[Union[str, os.PathLike[str]]], Union[str, os.PathLike[str]]]) → Callable[[Union[str, os.PathLike[str]]], Union[str, os.PathLike[str]]]

Return a decorator to wrap functions that takes as inputs file to be downloaded from the remote source.

Parameters

f – function to be decorated, with signature f(filename)

Returns

wrapped function

pathTo(filename: Union[str, os.PathLike[str]]) → Union[str, os.PathLike[str]]

Generate local path.

Parameters

filename – str

Returns

path to file

upload(filename: Union[str, os.PathLike[str]]) → requests.models.Response

Upload the file to the given url.

Parameters

filename – name of the file to be uploaded

Returns

requests.Response

class cgnal.core.utils.cloud.HTTPRequestHandler(*args, directory=None, **kwargs)

Bases: cgnal.core.logging.defaults.WithLogging, http.server.SimpleHTTPRequestHandler

Performs POST operation.

do_POST() → None

Post given resource.

cgnal.core.utils.decorators module

Module that gathers useful decorators.

class cgnal.core.utils.decorators.Cached

Bases: object

Class to cache results and export them as pickle to be later reloaded.

static cache(func: Callable[[cgnal.core.utils.decorators.Cached], cgnal.core.typing.T]) → property

Return a decorator to cache function return values.

Parameters

func – input function

Returns

function wrapper

clear_cache() → None

Clear cache of the object.

load(filename: Union[str, os.PathLike[str]]) → None

Load pickle at given path (or all pickles in given folder).

Parameters

filename – path to pickles

classmethod load_element(filename: Union[str, os.PathLike[str]], prefix: str = '') → Iterable[Tuple[Union[str, os.PathLike[str]], Any]]

Load given element/property of the object.

Parameters

  • filename – name of the file/directory where the data should be taken from

  • prefix – prefix to add to elements in filename in case it is a directory

Yield

iterable of couples (path, loaded object)

static save_element(filename: Union[str, os.PathLike[str]], obj: Any) → None

Save given object in given file.

Parameters

  • filename – saving path

  • obj – object to be saved

Raises

Exception – if any error occurs while saving the file

save_pickles(path: Union[str, os.PathLike[str]]) → None

Save pickle in given path.

Parameters

path – saving path

cgnal.core.utils.decorators.cache(func: Callable[[...], cgnal.core.typing.T]) → Callable[[...], cgnal.core.typing.T]

Return a decorator to cache function return values.

Parameters

func – input function

Returns

function wrapper

cgnal.core.utils.decorators.lazyproperty(obj: Any) → property

Return a lazy property, i.e. a property that is computed if needed and then cached.

Parameters

obj – method do decorate

Returns

wrapped method

cgnal.core.utils.decorators.paramCheck(function: Callable[[...], cgnal.core.typing.T], allow_none: bool = True) → Callable[[...], cgnal.core.typing.T]

Return a decorator that performs runtime checks on the input types.

Parameters

  • function – function to be checked against its typing annotations

  • allow_none – whether None values are allowed

Returns

wrapped function

cgnal.core.utils.decorators.param_check(with_none: bool) → Callable[[...], Any]

Return a decorator that performs runtime checks on the typing of the inputs.

Parameters

with_none – whether None values are allowed

Returns

decorator

cgnal.core.utils.dict module

Module for basic dictionary functionalities.

cgnal.core.utils.dict.filterNones(_dict: Dict[cgnal.core.typing.T, Any]) → Dict[cgnal.core.typing.T, Any]

Return a dictionary where the key,value pairs are filtered where the value is None.

Parameters

_dict – dict with Nones

Returns

dict without Nones

cgnal.core.utils.dict.flattenKeys(input_dict: Dict[str, Any], sep: str = '.') → Dict[str, Any]

Flatten dictionary keys.

Return a dictionary whose nested keys have been flattened into the root level recursively flattening a multilevel nested dict into a single level dict with flattened keys and corresponding values. The keys are joined by the given separator.

Parameters

  • input_dict – a multilevel dict ex {“a”: {“b”: {“c”: 2}}, “d”: 2, “e”: 3}

  • sep – str delimiter to join keys

Returns

dict with flattened keys

cgnal.core.utils.dict.groupBy(lst: Iterable[cgnal.core.typing.T], key: Callable[[cgnal.core.typing.T], cgnal.core.typing.SupportsLessThan]) → Iterator[Tuple[cgnal.core.typing.SupportsLessThan, List[cgnal.core.typing.T]]]

Perform groupBy operation on a list according to the given key.

The function uses itertools groupby but on a sorted list.

Parameters

  • lst – List

  • key – function to be used as a key to perform groupby operation

Yield

Iterator

cgnal.core.utils.dict.groupIterable(iterable: Iterable[cgnal.core.typing.T], batch_size: int = 10000) → Iterator[List[cgnal.core.typing.T]]

Split a given iterable into batches of given batch_size.

Parameters

  • iterable – iterable

  • batch_size – int

Returns

Iterator

cgnal.core.utils.dict.pairwise(iterable: Iterable[cgnal.core.typing.T]) → zip

Return a pairing of elements of an iterable.

Example: s -> (s0,s1), (s1,s2), (s2, s3), …

Parameters

iterable – iterable whose elements ought to be paired

Returns

zipped iterable

cgnal.core.utils.dict.unflattenKeys(input_dict: Dict[str, Any], sep: str = '.') → Dict[str, Any]

Transform a dict into a nested dict by splitting keys on the given separator.

A dict with key as {‘a.b’:2} will be transformed into a dict of dicts as {‘a’:{‘b’:2}}

Parameters

  • input_dict – dict

  • sep – str delimiter to split keys

Returns

dict

cgnal.core.utils.dict.union(*dicts: dict) → dict

Return a dictionary that results from the recursive merge of the input dictionaries.

Parameters

dicts – list of dicts

Returns

merged dict

cgnal.core.utils.email module

Basic functionalities for interacting with email servers.

class cgnal.core.utils.email.EmailSender(email_address: str, username: str, password: str, smtp_address: str, auth_protocol: str = 'None', port: Optional[int] = None)

Bases: cgnal.core.logging.defaults.WithLogging

Email Sender Utility Class.

Return the instance of the class that acts as client to interact with an email server.

Parameters

  • email_address – Sender email address

  • username – Username for authentication

  • password – Password for authentication

  • smtp_address – SMTP server address

  • auth_protocol – Authentication protocol to use

  • port – Port of SMTP server

send_mail(text: str, subject: str, destination: str, attachments: Optional[List[str]] = None) → None

Send email.

Parameters

  • text – The text of the email

  • subject – The subject of the email

  • destination – The destination email address

  • attachments – List with the files to send as attachments to the email

Raises

NotImplementedError – if given protocol is not implemented

cgnal.core.utils.fs module

Module for basic functionalities to operate with local file systems.

cgnal.core.utils.fs.create_dir_if_not_exists(directory: Union[str, os.PathLike[str]]) → Union[str, os.PathLike[str]]

Create a directory if it does not exist.

Parameters

directory – path

Returns

directory, str

cgnal.core.utils.fs.get_lexicographic_dirname(dirpath: Union[str, os.PathLike[str]], first: bool = False) → Union[str, os.PathLike[str]]

Return the first (or last) subdirectory name ordered lexicographically.

Parameters

  • dirpath – name of the base path whose subdirectories ought to be listed

  • first – whether the first or the last element should be returned

Returns

first (or last) subdirectory name ordered lexicographically.

cgnal.core.utils.fs.mkdir(path: Union[str, os.PathLike[str]]) → None

Create a dir, using a formulation consistent between 2.x and 3.x python versions.

Parameters

path – path to create

Raises

OSError – whenever OSError is raised by makedirs and it’s not because the directory exists

cgnal.core.utils.pandas module

Basic utilities extending pandas functionalities.

cgnal.core.utils.pandas.is_sparse(df: pandas.core.frame.DataFrame) → bool

Return True if a dataframe is sparse.

Parameters

df – pandas dataframe

Returns

bool

cgnal.core.utils.pandas.loc(df: pandas.core.frame.DataFrame, idx: Union[pandas.core.indexes.base.Index, List[Any]])

Return a filtered dataframe based on an index list and it is designed to work propertly with sparse dataframe as well.

Parameters

  • df – DataFrame

  • idx – index list

Returns

DataFrame which is filtered

Module contents

Generic utilities.