modutils 0.2.0

A library with modern utilities to assist development efficiency

Modutils

Modern python3 utilities
A collection of classes and functions

Skip to docs

Most common utilities

aioloop

This function provides an easy way to use another defined function to provide async functionality. Named arguments should be provided in a single dictionary inside the list of arguments.

import requests
from modutils import aioloop

# example w/o named arguments
def add(x:int,y:int)->int:return x+y

args = [[x,y] for x in range(0,5) for y in range(5,10)]
list_of_returns = aioloop(add, args)


# example w/ named arguments
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?'}}],
        ['https://www.github.com']
    ]
list_of_returns = aioloop(get_url, args)

aiobulk

aiobulk is a decorator used to add a bulk attribute to the existing function. This decorator leverages aioloop for async functionality.

import requests
from modutils.decorators import aiobulk

# example w/o named arguments
@aiobulk
def add(x:int,y:int)->int:return x+y

args = [[x,y] for x in range(0,5) for y in range(5,10)]
list_of_returns = add.bulk(args)


# example w/ named arguments
@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?'}}],
        ['https://www.github.com']
    ]
list_of_returns = get_url.bulk(args)

BaseSession

This class is a modified requests session class that enables logging per request and will persistently attempt to resolve requests with incorrect status codes.

from modutils.http import BaseSession

BaseAsyncSession

This class is built with BaseSession and aiobulk to provide a bulk attribute for each request functions: get, put, post, patch, head, and delete

from modutils.http import BaseAsyncSession
session = BaseAsyncSession()
args = [
        ['https://www.google.com', {'params':{'q':'Why is the sky blue?'}}],
        ['https://www.github.com']
    ]
list_of_returns = session.get.bulk(args)

Email

The Email class allows a simple way to send emails via an non-authenticated or authenticated session

from modutils.http import Email

Urlscraper

The urlscraper method is an easy way to search web pages for matching content of a given pattern

from modutils.http import urlscraper
# this example will use a regex pattern to find all the sha256 hashes in a given blog
# note: the blog is fake and will need to be replaced for example
sha256hashes = urlscraper('https://fakeblogsite.com', '[A-Fa-f0-9]{64}', regex=True)

echo

echo is a fancy print that allows you to easily print objects in their prettiest form, add color, and flush the current line.

from modutils import echo
echo({'hello': 'world'}, color='red')

Documentation

• modutils.aio
  • aioloop
• modutils.decorators
  • aiobulk.__init__
  • aiobulk.bulk
• modutils.http
  • urlscraper
  • modutils.http.BaseAsyncSession
    • BaseAsyncSession.__init__
  • modutils.http.BaseSession
    • BaseSession.__init__
  • modutils.http.Email
    • Email.__init__
    • Email.send
• modutils
  • echo
  • globpath
  • has_package
  • import_from
  • install_package
  • list_packages
  • nget
  • response_to_str
  • scroll
  • update_package
  • modutils.sha256

modutils.aio

modutils.aio defines a new type called Eventloop. This type is created at runtime to specify the type by operating system.

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.decorators

aiobulk.__init__(self, function: Callable)

initialize aiobulk the function being decorated will receive a new attribute for the bulk call

    :param function: function being decorated

aiobulk.bulk(self, args_list: List[list], loop: <function NewType.<locals>.new_type at 0x0000021B97E921F0> = 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

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

        :return list of results

modutils.http

urlscraper(url: str, pattern: str, regex: bool = False) -> list

urlscraper is a simple method to scrape information from a url based on a given string pattern

:param url: the url to run pattern against
:param pattern: the string representation of the pattern
:param regex: flag for using a pattern as regex or string compare

:return: list of strings that matched or contained pattern

modutils.http.BaseAsyncSession(self, *args, **kwargs)

BaseAsyncSession to be used for sessions that need to be able to make asynchronous requests

BaseAsyncSession.__init__(self, *args, **kwargs)

initialize BaseAsyncSession

    :param args: list of args
    :param kwargs: dict of named args

modutils.http.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)

BaseSession that will log requests and persist requests to attempt to resolve desired 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 BaseSession

    :param max_retries: maximum amount of retries if non resolved status code found
    :param pool_connections: number of pool connection; default 16
    :param pool_maxsize: max number of connections in pool; default 16
    :param resolve_status_codes: extra status codes to resolve; default None
    :param verbose: more verbose logging output if response fails; default False
    :param auth: basic auth username and password tuple; default None        

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

Class to easily send emails

Email.__init__(self, smtp_server, smtp_port, from_address: str = None, authentication_required: bool = False, auth_username: str = None, auth_password: str = None)

Create a new email session

Email.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

modutils

echo(content: Any, list_delimiter: str = '\n', indent: int = 4, color: str = None, flush: bool = False) -> None

echo - automatically pretty print or resolve to a printable object

:param content: the object to print
:param list_delimiter: delimiter to join list; default: 

:param indent: indent space count; default: 4
:param color: change color of text; default: None
:param flush: flush will make the current line be overwritten; default: False
:return: None

globpath(filepath: str) -> str

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

filepath {str} -- filepath to glob

return {str} -- new globpath

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

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

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

nget(dictionary: dict, keys: Iterable, default: Any = None) -> 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.stdutils

scroll(content: list, sleep_timer=1, list_delimiter: str = '\n', indent: int = 4, color: str = None) -> None

use echo and a delay to scroll text over same line

:param content: the object to print
:param list_delimiter: delimiter to join list; default: 

:param indent: indent space count; default: 4
:param color: change color of text; default: None
:return: None

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.sha256(self, value)

sha256 type