Python/Bash utils by qianws and his collections
--- A util package for human (usability/readability is the first-concern)
Basically, it's a warehouse of all the handy "wheels" I built/collected along my life as a Pythonista.
Just open and use, intuitively as the following examples:
start_in_thread(fn, *args, **kwargs)single line to run a function in background threadauto_unstub()Automatically unstub mockito patch/mocksmockifya context protocol for modifying one or more objects
from qPyUtils.debug import auto_unstub, mockify, start_in_thread
# ---------- start_in_thread
start_in_thread(function_to_start_a_web_server, host='localhost', port=8011)
time.sleep(0.1) # waiting server to start, critical for Travis-CI
assert requests.get('http://localhost:8011/...') == ...
# ---------- auto_unstub
@auto_unstub # `mockito.unstub()` is inserted into `tearDown()` logic
class TestInitLog(TestCase):
def test_method(self):
... # code that may use mockito
def test_method2(self):
# ---------- mockify
with mockify(sys.stderr) as sys.stderr: # temporarily suppress stderr
... # code that calls `sys.stderr.write()`
verify(sys.stderr, atleast=1).write(Ellipsis)Just inherent the BaseLogParser and implement some methods;
the framework will do the heavy-lifting for you, including:
- filtering files by date
- using thread/process_pool for multiple loading files in parallel
- logging malformed lines
- recursively find and load all target files under specified path
from qPyUtils.log.parser.base import BaseLogParser
# extend and implement some logic (please read full API documentation for complete info)
class MyLogLoader(BaseLogParser):
LOG_TYPE = 'my_log'
def glob_files(self, base_path):
# type: (Path) -> Iterable[Path]
return base_path.rglob('*.log')
def logfile2blocks(self, path):
# type: (Path) -> Iterable[Text]
from qPyUtils.text import csplit # python mimic for the `csplit` command in bash
with path.open(mode='rt', encoding='utf8') as lines:
return (''.join(block) for block in csplit(lines, re.compile(r'^{"version')))
def block2records(self, block):
# type: (Text) -> Iterable[dict]
for rec in dirty_json_or_none(block)['records']:
yield dict(**rec)
# now, use the magic power for free
my_log_loader = MyLogLoader()
my_log_loader.load_dir('./log')
my_log_loader.load_single_file('./log/my_log.2018-08-01.log')The easy & standardized usage of logger in Python
from qPyUtils.log.writer import init_log
# set logger_name to None or ''(as default), to use the root-logger
# the directories along the log_path will be automatically created if not yet exist
# a reasonable default format string, and daily log rotation is also provided.
logger = init_log(log_path='/home/my_name/my_project/log/my_log_file')
# set log_path to None or ''(as default)
# so that no file handlers are involved, only logs to console
logger = init_log()
# set `is_writing_console=False` to suppress console output
# not that for such case, you must provide a log_path; otherwise, no handlers will work
logger = init_log(is_writing_console=False, log_path='./log/my_log_file')import time
import qPyUtils.log.timer
# user-defined output_fn; feel free to try using `output_fn=logger.info` etc.
res = []
with qPyUtils.log.timer.Timer('Long task 中文', output_fn=res.append) as timer:
with timer.child('large step'):
time.sleep(1)
for _ in range(5):
with timer.child('small step'):
time.sleep(0.5)
print(res[0])
# ----- result:
# Long task 中文: 3.506s
# 5x small step: 2.503s (71%)
# 1x large step: 1.001s (28%)
# user-defined format_string
with qPyUtils.log.timer.Timer('Long task 中文', fmt='{name} --> {elapsed:.3f}') as timer:
time.sleep(0.1)
# ----- result:
# Long task 中文 --> 0.101sdummy_fn(*args, **kwargs): accepts anything but does nothingidentify_fn(arg): just reflect theargpassed inT: atyping.TypeVarfor generic programming type-hints
A light weight thread/process pool
- intuitive interface
- auto utilize multi CPU cores
- progress bar
- support
*args/*kwargs - Exception object returned in place of the corresponding task result
from qPyUtils.parallel import para
import math
# simple usage; pass in tasks and function, get results
tasks = range(5)
fn = math.factorial
assert [1, 1, 2, 6, 24] == para(tasks, fn)
# you can still degrade to single thread --- sequential
# or even suppress the progress bar
para(tasks, fn, n_jobs=1, is_suppress_progressbar=True)
# support kwargs via the `use_kwargs` param
# CAUTION: `fn` should avoid appear in closure of the `para()` func; this is just for demo
fn2 = lambda x: math.factorial(x)
para([{'x': t} for t in tasks], fn2, use_kwargs=True)Data streaming / Functional programming related utils.
from qPyUtils.streaming import Repeat
# ---------- decorator over function
# turns a generator factory function into a sequence-like iterable;
# which could be iterated multiple epochs
@Repeat
def my_gen():
for i in range(3):
yield i
# NOTE: now the name `my_gen` is the wrapped structure; no parenthesis here
assert (0, 1, 2) == tuple(my_gen)
assert (0, 1, 2) == tuple(my_gen)
# ---------- decorator over method
class MyClazz(object):
@Repeat
def my_method(self, a, b, prefix='>>>'):
for i in range(a, b):
yield '{}{}'.format(prefix, i)
obj = MyClazz()
my_gen = obj.my_method(0, 3, prefix=':') # call as normal
assert (':0', ':1', ':2') == tuple(my_gen)
assert (':0', ':1', ':2') == tuple(my_gen)from qPyUtils.text import *
>>> assert is_none_or_empty('')
>>> assert is_none_or_empty(None)
# avoid python2's stupid handling logic for str/unicode
>>> dump_utf8({'name': u'中文'}) # use json.dumps() to avoid the ugly `\u....` / `\x..`
{"name": "中文"}
>>> lines = [
... 'line1',
... '---- line2',
... 'line3',
... 'line4',
... '---- line5',
... 'line6'
... ]
>>> pattern = r'----'
# mimic the `csplit` command in bash
# which groups text lines into blocks; from current line (inclusive) to next pattern-matching-line (exclusive)
# auto handles last buffer or empty input sequence
>>> list(csplit(lines, pattern))
[('line1',), ('---- line2', 'line3', 'line4'), ('---- line5', 'line6')]
# unify str/unicode/bytes in py2/py3 into Text type, using UTF8
>>> ensure_text(b'abc')
u'abc'
>>> ensure_text(b'中文')
u'中文'
>>> ensure_text(u'中文')
u'中文'One-liner annotation turns your function into a RESTful service
# ----- server side (CAUTION: UNICODE_LITERALS OR u'...' IS NECESSARY FOR NON-ASCII REQUESTS)
# encoding: utf-8
from __future__ import unicode_literals
from qPyUtils.web import RESTful
@RESTful(port=8004, route='/')
def introduce(name, friends):
return '{} has friends: {}'.format(name.upper(), ', '.join(friends))
introduce.serve()# ----- client side
# Unicode is automatically translated by UTF8
$ curl 'http://localhost:8004?name=koyo' -d 'friends=solar' -d 'friends=ape' -d 'friends=tutor' -d 'friends=斑马'
KOYO has friends: solar, ape, tutor, 斑马%A command line tool for forwarding port(s); Unicode Domain-Name is also supported.
# ----- ensure PYTHON_BIN is in the PATH; consider adding it to ~/.bashrc
PYTHON_BIN=$(python -c 'from distutils.sysconfig import EXEC_PREFIX as p; print(p + "/bin")')
export PATH=${PYTHON_BIN}:$PATH
portforward -H www.nic.ad.jp -p 80 -l 8012
# ----- OR just simply call by `python -m ...` as an ad-hoc solution
python -m qPyUtils.system.portforward -H www.nic.ad.jp -p 80 -l 8012Usage:
portforward (-c <conf_path> | -H <host> -p <port> -l <local_port>)
portforward -h | --help
Demo conf file:
<host> <port> <local_port>
www.nic.ad.jp 80 8012
中国互联网络信息中心.中国 80 8013
A function simulating rm -f <path>
- support both dir and file
- ignore_errors, if not exist
from qPyUtils.system.file_system import rm
rm('path/to/your/dir/or/file', ignore_errors=True)- A *NIX OS
- python >= 2.7 or python >= 3.5
For common package users,
it's as simple as typing pip install qPyUtils and having a cup of coffee, then it's done.
If you want to develop it, please follow the instructions below:
git clone <url-of-this-repo> && cd qPyUtils/, get the repo and cd insidepip install -r requirements-dev.txt, install all the dependencies for developingpyb install_dependencies analyze -v, using the powerful PyBuilder build-system to get the rest job done.
- following the installing procedure for developers as above
pyb run_integration_tests -v
Read the .travis.yml file for deployment settings, I use Travis-CI for PyPI/ReadTheDocs continuous delivery.
- PyBuilder - The modern build tool for Python(just like maven/gradle for java)
- Travis-CI - The famous CI service vendor
- Sphinx - The de facto standard for docs the Python world
Bug, Issues, Doc, Pull Request ... --- Any contribution is welcomed! Feel free to get your hands dirty and help me improve it.
Most of the code is well-documented and carefully-tested; It's easier than you might think to extend it.
- QIAN Weishuo - Initial work - Facebook
See also the list of contributors who participated in this project.
This project is licensed under the MIT License
debug.auto_unstub()https://gist.github.com/jnape/5767029log.writer()http://styleguide.baidu.com/style/python/index.htmltext.dirty_json_or_none()https://github.com/codecobblers/dirtyjsonlog.timer.Timer()https://github.com/mherrmann/timer-cm