Decorator for logging function arguments and return value by human-readable way
Project description
threaded
threaded is a set of decorators, which wrap functions in:
concurrent.futures.ThreadPool
threading.Thread
asyncio.Task in Python 3.
gevent.threadpool.ThreadPool if gevent is installed.
Why? Because copy-paste of loop.create_task, threading.Thread and thread_pool.submit is boring, especially if target functions is used by this way only.
Pros:
Free software: Apache license
Open Source: https://github.com/penguinolog/threaded
PyPI packaged: https://pypi-hypernode.com/pypi/threaded
Tested: see bages on top
Support multiple Python versions:
Python 2.7 Python 3.4 Python 3.5 Python 3.6 PyPy PyPy3 3.5+ Jyton 2.7
Decorators:
ThreadPooled - native concurrent.futures.ThreadPool usage on Python 3 and it’s backport on Python 2.7.
threadpooled is alias for ThreadPooled.
Threaded - wrap in threading.Thread.
threaded is alias for Threaded.
AsyncIOTask - wrap in asyncio.Task. Uses the same API, as Python 3 ThreadPooled.
asynciotask is alias for AsyncIOTask.
GThreadPooled - wrap function in gevent.threadpool.ThreadPool.
gthreadpooled is alias for GThreadPooled.
Usage
ThreadPooled
Mostly it is required decorator: submit function to ThreadPoolExecutor on call.
threaded.ThreadPooled.configure(max_workers=3)Python 2.7 usage:
@threaded.ThreadPooled
def func():
pass
concurrent.futures.wait([func()])Python 3.3+ usage:
@threaded.ThreadPooled
def func():
pass
concurrent.futures.wait([func()])Python 3.3+ usage with asyncio:
loop = asyncio.get_event_loop()
@threaded.ThreadPooled(loop_getter=loop, loop_getter_need_context=False)
def func():
pass
loop.run_until_complete(asyncio.wait_for(func(), timeout))Python 3.3+ usage with asyncio and loop extraction from call arguments:
loop_getter = lambda tgt_loop: tgt_loop
@threaded.ThreadPooled(loop_getter=loop_getter, loop_getter_need_context=True) # loop_getter_need_context is required
def func(*args, **kwargs):
pass
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait_for(func(loop), timeout))During application shutdown, pool can be stopped (while it will be recreated automatically, if some component will request).
threaded.ThreadPooled.shutdown()Threaded
Classic threading.Thread. Useful for running until close and self-closing threads without return.
Usage example:
@threaded.Threaded
def func(*args, **kwargs):
pass
thread = func()
thread.start()
thread.join()Without arguments, thread name will use pattern: ‘Threaded: ‘ + func.__name__
Override name can be don via corresponding argument:
@threaded.Threaded(name='Function in thread')
def func(*args, **kwargs):
passThread can be daemonized automatically:
@threaded.Threaded(daemon=True)
def func(*args, **kwargs):
passAlso, if no any addition manipulations expected before thread start, it can be started automatically before return:
@threaded.Threaded(started=True)
def func(*args, **kwargs):
passAsyncIOTask
Wrap in asyncio.Task.
usage with asyncio:
@threaded.AsyncIOTask
def func():
pass
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait_for(func(), timeout))Provide event loop directly:
loop = asyncio.get_event_loop()
@threaded.AsyncIOTask(loop_getter=loop)
def func():
pass
loop.run_until_complete(asyncio.wait_for(func(), timeout))Usage with loop extraction from call arguments:
loop_getter = lambda tgt_loop: tgt_loop
@threaded.AsyncIOTask(loop_getter=loop_getter, loop_getter_need_context=True)
def func(*args, **kwargs):
pass
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait_for(func(loop), timeout))GThreadPooled
Post function to gevent.threadpool.ThreadPool.
threaded.GThreadPooled.configure(max_workers=3)Basic usage example:
@threaded.GThreadPooled
def func():
pass
func().wait()Testing
The main test mechanism for the package threaded is using tox. Test environments available:
pep8 py27 py34 py35 py36 pypy pypy3 pylint
CI systems
For code checking several CI systems is used in parallel:
Travis CI: is used for checking: PEP8, pylint, bandit, installation possibility and unit tests. Also it’s publishes coverage on coveralls.
coveralls: is used for coverage display.
CD system
Travis CI: is used for package delivery on PyPI.
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
Built Distribution
File details
Details for the file threaded-0.6.0.tar.gz.
File metadata
- Download URL: threaded-0.6.0.tar.gz
- Upload date:
- Size: 11.4 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 | b79ce72df3604111789cea1b336e300f347346d6ecbc974879083eaffc456cd0 |
|
| MD5 | 8511d5ff2a12f6eb9b5717a241d2085d |
|
| BLAKE2b-256 | 081ccdfa792f6e1e2e113d672f0467af00d682a90f1b912d056461ad4582a2a3 |
File details
Details for the file threaded-0.6.0-py2.py3-none-any.whl.
File metadata
- Download URL: threaded-0.6.0-py2.py3-none-any.whl
- Upload date:
- Size: 18.9 kB
- Tags: Python 2, Python 3
- Uploaded using Trusted Publishing? No
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 | f53a62238dc9368aafd4e19a1e5b54f20c8433ee2c203a2642765a869597f343 |
|
| MD5 | fe130d17ab47c5f58ab802bad93316be |
|
| BLAKE2b-256 | d40acb90a17ac42bc4d2b801ee51830e5aad500382c152ea642d6fb9e9f77406 |
