modutils 0.1.4

A library with modern utilities to assist development efficiency

Modutils

A library with modern utilities to assist development efficiency 

Documentation

• modutils.aioutils
  • aiobulk
  • aioloop
• modutils.hashutils
  • calc_sha256_from_dir
  • calc_sha256_from_dir_map
  • sha256_from_dir
  • sha256_from_dir_map
  • sha256_from_file
• modutils.importutils
  • globpath
  • import_from
• modutils.packageutils
  • has_package
  • install_package
  • list_packages
  • update_package
• modutils.sessionutils
  • modutils.sessionutils.BaseSession
    • BaseSession.init
    • BaseSession._log_response
    • BaseSession._resolver
    • BaseSession.delete
    • BaseSession.get
    • BaseSession.head
    • BaseSession.patch
    • BaseSession.post
    • BaseSession.put
  • modutils.sessionutils.EmailSession
    • EmailSession.enter
    • EmailSession.exit
    • EmailSession.init
    • EmailSession.send
• modutils.stdutils
  • modutils.stdutils.echo
    • echo.call
    • echo.init
    • echo._resolve
    • echo._resolve_class_object
    • echo._resolve_response_object
• modutils.typeutils
  • nget
  • modutils.typeutils.sha256

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