Skip to main content

A library with modern utilities to assist development efficiency

Project description

Modutils

A library with modern utilities to assist development efficiency 

Documentation

modutils.aioutils

aiobulk(function, loop: Eventloop = None, max_async_pool: int = 16, max_futures: int = 100000, disable_progress_bar: bool = False, progress_bar_color: str = 'green_3a', progress_bar_format: str = None)

add a method called 'bulk' to given function

    :param function {Callable}: function to map to arguments
    :param loop {Eventloop}: a pre-defined asyncio loop
    :param max_async_pool {int}: max async pool, this will define the number of processes to run at once
    :param max_futures {int}: max futures, this will define the number of processes to setup and execute at once.
        If there is a lot of arguments and futures is very large, can cause memory issues.
    :param disable_progress_bar {bool}: disable progress bar from printing
    :param progress_bar_color {str}: color of progress bar; default: green
    :param progress_bar_format {str}: format for progress bar output; default: None

    Examples:
        @aiobulk
        def add(x:int,y:int)->int:return x+y


        # call the original function
        add.add(1,2)

        # call the newly added bulk function
        args = [[x,y] for x in range(0,5) for y in range(5,10)]
        list_of_returns = add.bulk(args)


        @aiobulk
        def get_url(url, params:dict=None):
            return requests.get(url, params=params)

        args = [
            ['https://www.google.com', {'params':{'q':'Why is the sky blue?'}}]
        ]
        responses = get_url.bulk(args)


    :return list of results

aioloop(function: Callable, args_list: List[List], loop: Eventloop = None, max_async_pool: int = 16, max_futures: int = 100000, disable_progress_bar: bool = False, progress_bar_color: str = 'green_3a', progress_bar_format: str = None) -> list

create new aioloop, run, and return results

:param fn {Callable}: function to map to arguments
:param args_list {List[List]}: list of arguments to send to function
:param loop {Eventloop}: a pre-defined asyncio loop
:param max_async_pool {int}: max async pool, this will define the number of processes to run at once
:param max_futures {int}: max futures, this will define the number of processes to setup and execute at once.
    If there is a lot of arguments and futures is very large, can cause memory issues.
:param disable_progress_bar {bool}: disable progress bar from printing
:param progress_bar_color {str}: color of progress bar; default: green
:param progress_bar_format {str}: format for progress bar output; default: None


:return list of results

modutils.hashutils

calc_sha256_from_dir(directory: str) -> list

calculate sha256 values from files found in a directory

directory {str} -- path to directory to search

return {list} -- list of sha256 values found

calc_sha256_from_dir_map(directory: str) -> dict

calculate sha256 values from files found in a directory and map them to their filepath

directory {str} -- path to directory to search

return {dict} -- dictionary with filepath as key and sha256 as value

sha256_from_dir(directory: str) -> list

extract all sha256 values from all files inside of a directory using regex

directory {str} -- path to directory to search

return {list} -- list of sha256 values found

sha256_from_dir_map(directory: str) -> dict

extract all sha256 values from all files inside of a directory using regex and map them to their filepath

directory {str} -- path to directory to search

return {dict} -- dictionary of filepath as key and found sha256 list as value

sha256_from_file(filepath: str) -> list

extract all sha256 values from a file using regex

filepath {str} -- path to file to search

return {list} -- list of sha256 values found

modutils.importutils

globpath(filepath: str) -> str

convert a filepath to a glob path, ['/'|''] to '.'

filepath {str} -- filepath to glob

return {str} -- new globpath

import_from(globpath: str, name: str) -> Union[object, NoneType]

Import module and return instance of given function name

globpath {str} -- the filepath in glob format
name {str} -- the method name to import

return {Union[object, None]} -- method attribute of import

modutils.packageutils

has_package(name: str, version: Union[str, int, float] = None) -> bool

check if current environment has a package

name {str} -- name of package to check
version {Union[str, int, float]} -- OPTIONAL, will append version for a specific version check of package

return {bool} -- True if package was found and False if not

install_package(name: str, force: bool = False, extra_index: str = None, trusted_host: str = None) -> Tuple[str, str]

install a pip3 package using a subprocess in current environment

name {str} -- name of package to install
force {bool} -- force newest edition
extra_index {str} -- extra url to index in package manager
trusted_host {str} -- extra url where package is hosted

return {tuple} -- return the output, and error of subprocess run

list_packages() -> list

list current pip3 packages in environment

return {list} -- a list of available packages

update_package(name: str, extra_index: str = None, trusted_host: str = None) -> Tuple[str, str]

update a pip3 package by name, leverages install_package with force = True

name {str} -- name of package to install
extra_index {str} -- extra url to index in package manager
trusted_host {str} -- extra url where package is hosted

return {tuple} -- return the output, and error of subprocess run

modutils.sessionutils

modutils.sessionutils.BaseSession(self, max_retries: int = 3, pool_connections: int = 16, pool_maxsize: int = 16, resolve_status_codes: list = None, verbose: bool = False, auth: tuple = None)

A base session to build persistent sessions to handle resolving specific status codes

BaseSession.__init__(self, max_retries: int = 3, pool_connections: int = 16, pool_maxsize: int = 16, resolve_status_codes: list = None, verbose: bool = False, auth: tuple = None)

Initialize a new session

BaseSession._log_response(self, response: requests.models.Response) -> None

log each requests from resolver

BaseSession._resolver(self, request: functools.partial) -> requests.models.Response

attempt to resolve a requests with an invalid status code

    if the status code of the requests is not one to resolve:
        Default:  [200, 201, 202, 203, 204, 205, 206, 207, 208, 226]
    the requests will be sent up to max_retries or until receiving an accepted status_code

    :param request: partial requests function to be used to attempt and resolve a valid response

    :return: response from the requests

BaseSession.delete(self, url: Union[str, bytes], **kwargs) -> requests.models.Response

override of requests.Session delete calls Session.get with resolver

    :param url: url for requests
    :param kwargs: named arguments for requests

    :return: response from requests

BaseSession.get(self, url: Union[str, bytes], **kwargs) -> requests.models.Response

override of requests.Session get calls Session.get with resolver

    :param url: url for requests
    :param kwargs: named arguments for requests

    :return: response from requests

BaseSession.head(self, url: Union[str, bytes], **kwargs) -> requests.models.Response

override of requests.Session head calls Session.get with resolver

    :param url: url for requests
    :param kwargs: named arguments for requests

    :return: response from requests

BaseSession.patch(self, url: Union[str, bytes], **kwargs) -> requests.models.Response

override of requests.Session patch calls Session.get with resolver

    :param url: url for requests
    :param kwargs: named arguments for requests

    :return: response from requests

BaseSession.post(self, url: Union[str, bytes], **kwargs) -> requests.models.Response

override of requests.Session post calls Session.get with resolver

    :param url: url for requests
    :param kwargs: named arguments for requests

    :return: response from requests

BaseSession.put(self, url: Union[str, bytes], **kwargs) -> requests.models.Response

override of requests.Session put calls Session.get with resolver

    :param url: url for requests
    :param kwargs: named arguments for requests

    :return: response from requests

modutils.sessionutils.EmailSession(self, smtp_server, smtp_port, from_address: str = None, authentication_required: bool = False, auth_username: str = None, auth_password: str = None)

New email session

EmailSession.send(self, subject: str, body: str, to_address_list: list, cc_address_list: list = None, from_address: str = None, encoding: str = 'html', logo_images: list = None, file_attachments: list = None) -> dict

:param subject: Subject string for email, required
:param body: Message content for email, required
:param to_address_list: addresses to send email to, required
:param cc_address_list: addresses to cc on email, default: None
:param from_address: address to send email from, default: None, will use self.from_address if one was given
:param encoding: encoding for body, default: html
:param logo_images: list of paths to images to use for logos, default: None
:param file_attachments: list of paths to attachments for email, default: None

:return: dict

send email    

modutils.stdutils

fg(color)

alias for colored().foreground()

main(md=None, filename=None, cols=None, theme=None, c_theme=None, bg=None, c_no_guess=None, display_links=None, link_style=None, from_txt=None, do_html=None, code_hilite=None, c_def_lexer=None, theme_info=None, no_colors=None, tab_length=4, no_change_defenc=False, header_nrs=False, **kw)

md is markdown string. alternatively we use filename and read

modutils.stdutils.echo(self, message: Any, expand: bool = False, list_delimiter: str = '\n', indent: int = 4, color: str = None, newline: bool = True, markdown: bool = False)

echo - a modern print statement that resolves objects to printable types

echo.__call__(self, message: Any = None)

call can be used to overwrite the current message if you want to reuse an echo object

        x = echo('123')
        x()
        x('345')


    message {Any} -- object to print

echo.__init__(self, message: Any, expand: bool = False, list_delimiter: str = '\n', indent: int = 4, color: str = None, newline: bool = True, markdown: bool = False)

None

echo._resolve(self, message: Any) -> None

resolve and print a message based on Any given type

    message {Any} -- an object to print

    return {None} -- does not return anything, this function will write to stdout

echo._resolve_class_object(self, cls: object)

resolve a class object to an always printable form based on the type of attributes. recursively resolves attributes

    cls {object} -- class object to print

    return {None} - does not return anything due to needing to call resolve

echo._resolve_response_object(self, response: requests.models.Response) -> str

resolve requests.Response objects to an always printable form based on their status_code from response

    response {requests.Response} -- response object to resolve

    return {str} -- a printable form of the response

modutils.typeutils

nget(d: dict, *args: Union[str, list]) -> Any

nget - nested get call to easily retrieve nested information with a single call and set a default Ex. nget(dict, ['key1', 'key2', ..], default) nget(dict, key1, key2, .., default)

    nget use an iterable of keys to retrieve nested information and can set a default if a key is not found

modutils.typeutils.sha256(self, value)

An implementation of a sha256 type based off the str type

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

modutils-0.1.2.tar.gz (14.6 kB view hashes)

Uploaded Source

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page