Easy-to-use tools for handling function responses, handling exceptions, and creating asynchronous tasks.
Project description
ret-tools
ret-tools is a small library used for formating function responses, handling function exceptions, executing and handling asynchronous tasks. View examples below:
Handling Functions
ret-tools is built off two main functions: run()
and run_async()
both return a RetResp with four properties:
success (bool True or False), msg ("Success" or Exception), tb (None or Exception Traceback), resp (None or response of original function)
RetResp can be treated as
a list [success, msg, tb, resp]
,
a dict {"success": True, "msg": "Success", "tb": None, "resp": "some random data"}
,
or a str: Succeeded: some random data
,
Example of run()
:
import ret_tools as rt
def func(name):
return "Hi {}".format(name)
def main():
ret = rt.run(func, ("John Smith"))
print(ret.success)
# True
print(ret.msg)
# Success
print(ret.resp)
# "Hi John Smith"
print(tuple(ret))
# (True, "Success", None, "Hi John Smith")
if __name__ == "__main__":
main()
Example of run_async()
import ret_tools as rt
import asyncio
async def afunc(name):
await asyncio.sleep(1)
return "HI {}".format(name)
def main():
ret = asyncio.run(rt.run_async(afunc("John Smith")))
print(ret.success)
# True
print(ret.msg)
# Success
print(ret.resp)
# "Hi John Smith"
print(tuple(ret))
# (True, "Success", None, "Hi John Smith")
if __name__ == "__main__":
main()
If a function throws an Exception the response is as follows:
import ret_tools as rt
def func():
raise Exception("Something went wrong")
def main():
ret = rt.run(func)
print(ret.success)
# False
print(ret.msg)
# Exception("Something went wrong")
print(ret.resp)
# None
print(tuple(ret))
# (False, Exception("Something went wrong"), "Traceback: ...", None)
if __name__ == "__main__":
main()
ret-tools can be configured with a RetConfig object, you can specify a failed callback, error callback, and a task callback
failed callback ret_config.ff
gets called when a function throws an error, this can change on a function to function basis, you can also specify custom arguments, an example would be a websocket client that closes the connection when a send or recv fails
error callback ret_config.ec
gets called every time a function throws an error, it is global and can't change on a function-to-function basis, this can be used for logging purposes, the argument will be a RetResp
task callback ret_config.tc
gets called when a async task finishes, this can be changed on a task to task basis, the argument will be a RetResp
Example:
import ret_tools as rt
def error_callback(err_resp):
print(err_resp) # Failed: Exception Text
ret_config = rt.RetConfig(ec=error_callback)
rt.init(ret_config)
def failed(function_name, other_arg):
print("{} Failed, {}".format(function_name, other_arg))
# Do something if function fails
def func(*args):
raise Exception("Function Failed")
def main():
# function, fargs, failed_func, *failed_args
ret = rt.run(func, (), failed, "Name of function", "some other argument")
print(ret.success) # False
if __name__ == "__main__":
main()
Function Decorators
ret-tools offer two different function decorators: @def_ret()
and @async_ret()
each decorator will pass the function through rt.run()
or rt.run_async()
Example:
import ret_tools as rt
@rt.def_ret()
def func():
return "some random data"
def main():
ret = func()
print(ret.success) # True
print(ret.resp) # "some random data"
print(tuple(ret)) # (True, "Success", None, "some random data")
if __name__ == "__main__":
main()
You can also specify custom failed callbacks through the decorators
Example:
import ret_tools as rt
def function_failed(arg1, arg2):
print("Failed: {} {}".format(arg1, arg2)) # Failed: func failed
@rt.def_ret(function_failed, "func", "failed")
def func():
raise Exception("Failed")
def main():
ret = func()
print(ret.success) # True
print(ret.msg) # Failed
if __name__ == "__main__":
main()
Handling Tasks
ret-tools can also be used to handle asynchronous tasks, there are two main methods: create_task()
, wait_task()
Example of create_task()
:
import ret_tools as rt
import asyncio
def task_callback(resp): # response is a RetResp object
print(resp) # Succeeded: Hi John Smith
async def task(name):
await asyncio.sleep(1)
return "Hi {}".format(name)
def main():
t = rt.create_task(task("John Smith"), task_callback) # returns asyncio.Task
if __name__ == "__main__":
main()
create_task()
works even when no asyncio event loop is present, ret-tools will detect if a loop is running, if not it will create a new loop and run it in a seperate thread then it will execute the task from there
Example of wait_task()
:
import ret_tools as rt
import asyncio
async def task(name):
await asyncio.sleep(1)
return "Hi {}".format(name)
def main():
resp = rt.wait_task(task("John Smith"))
print(resp) # Succeeded: Hi John Smith
async def async_main():
resp = await rt.run_async(task("John Smith"))
print(resp) # Succeeded: Hi John Smith
resp = await rt.create_task(task("John Smith"))
print(resp) # Succeeded: Hi John Smith
if __name__ == "__main__":
main()
asyncio.run(async_main())
rt.wait_task()
is equivilent to await rt.create_task()
which is equivilent to await rt.run_async()
If there is no event loop use rt.wait_task()
, If you have an event loop and you want the response imediately use await rt.run_async()
, if you want the response in a callback use rt.create_task(task(), callback)
Good and Bad
RetResp's are built off two functions good()
and bad()
good()
returns (True, "Success", None)
always
bad()
returns (False, Exception, Traceback)
using sys.exc_info()
and traceback.print_exc()
Example:
import ret_tools as rt
def main():
ret = None
try:
print("Do Something Here")
ret = rt.good()
except:
ret = rt.bad()
print(tuple(ret)) # (True, "Success", None)
ret = None
try:
raise Exception("Something went wrong")
ret = rt.good() # Won't happen because exception is raised
except:
ret = rt.bad()
print(tuple(ret)) # (False, Exception("Something went wrong"), "Traceback: ...")
if __name__ == "__main__":
main()
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
Built Distribution
Hashes for ret_tools-1.0.4-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | cc692188ed02c3867738dd3476c2dcc7923d65aaa3fde0da1878f80e197837df |
|
MD5 | 235985f84b4465865ff6ec5f7b0b2d95 |
|
BLAKE2b-256 | 4e0a9ed54ba1391b280f212d37c0c0b22bd03e12696a23b5c9fa3a0c4adfb392 |