A library with modern utilities to assist development efficiency
Project description
Modutils
A library with modern utilities to assist development efficiency
Documentation
- modutils.aioutils
- modutils.hashutils
- modutils.importutils
- modutils.packageutils
- modutils.sessionutils
- modutils.stdutils
- modutils.typeutils
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
Release history Release notifications | RSS feed
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.3.tar.gz
(14.5 kB
view hashes)
