API Reference

conffor

sniputils.conffor.load(file: str)

load json file with default options

Parameters:file – file path to load Returns:json data as python object

sniputils.conffor.dump(file: str, config: dict, indent=2)

export dict as json file, with default options

Parameters:

• file – file path to export
• config – dict data object
• indent – json indent num

sniputils.conffor.read_list_csv(columns: List[str], file: str, **kwargs) → List[List]

read csv with columns, return raw data list

Returns:DataFrame rows value

sniputils.conffor.save_list_csv(data: List[List], columns: List[str], file: str, **kwargs)

store data list to csv

logsetting

sniputils.logsetting.reset_logbase(filename=None, stream=<_io.TextIOWrapper name='<stdout>' mode='w' encoding='UTF-8'>, filemode='a', level=20, msg_format='%(asctime)s <%(name)s: %(threadName)s> %(filename)s[line:%(lineno)d] %(levelname)s %(message)s', date_format='%Y-%m-%d %H:%M:%S')

Parameters:

• filename – which file will logging to
• stream – stream type, default is stdout
• filemode – [w]overwrite or [a]append mode, default is append
• level – logging level, default is INFO
• msg_format – message format pattern str, default is logsetting.MSG_FORMAT
• date_format – time format pattern str, default is logsetting.TIME_FORMAT

sniputils.logsetting.clear_logsetting()

clear all root logger handlers, reset level debug

cli

base on Click 6.x

Warning

notice that need LC_ALL=en_US.UTF-8 in sys env

in this package, its set in src.__init__ by locale.setlocale

use ops() os click.option() to add options

use @cli.resultcallback() to add callback function while parse cli params

Usage as typical example:

from sniputils.cli import cli, ops

cli.help = '''
    this is the typical example of sniputils.cli
'''

ops('--test', is_flag=True, default=False, help='use test model')

@cli.resultcallback()
def callback_parse_config(_, **options):
    print('cli parse callback!', options)
    parse_config(options)

$ python cli.py --test
# output -> 'cli parse callback!', {'test': True}

sniputils.cli.cli = <Group click cli>

this is a cli help docs which need be overwrite with assign cli.help attribute

use @cli.resultcallback() decorator to add callback function while parse cli params

class sniputils.cli.Separate(param_decls=None, show_default=False, prompt=False, confirmation_prompt=False, hide_input=False, is_flag=None, flag_value=None, multiple=False, count=False, allow_from_autoenv=True, type=None, help=None, hidden=False, show_choices=True, show_envvar=False, **attrs)

Usage: ops(cls=Separate, **kwargs)

option: '1,3,5,7,what,the,hell'

click params: ['1', '3', '5', '7', 'what', 'the', 'hell']

option: 'single'

click params: ['single']

class sniputils.cli.FlowRange(param_decls=None, show_default=False, prompt=False, confirmation_prompt=False, hide_input=False, is_flag=None, flag_value=None, multiple=False, count=False, allow_from_autoenv=True, type=None, help=None, hidden=False, show_choices=True, show_envvar=False, **attrs)

Usage: ops(cls=FlowRange, **kwargs)

option: 2-5

click params: [2, 3, 4, 5]

option: 1,2-5,3-6

click params: [1, 2, 3, 4, 5, 3, 4, 5, 6]

class sniputils.cli.TypeChose(*choices)

Usage: ops(type=TypeChose(click.IntParamType, click.File), **kwargs)

option: 34

click params: 34

option: './output.log'

click params: <_io.TextIOWrapper name='./output.log'>

Usage: ops(type=TypeChose(int, list), **kwargs)

option: ['what', 'the', 'hell']

click params: ['what', 'the', 'hell']

option: 12

click params: 12

database

concurrent

class sniputils.concurrent.Parallel(func: Callable[any, any], preload: Iterable = None, keep=False, progress=False, size=3)

Actor model extension

https://en.wikipedia.org/wiki/Actor_model

Usage:

# global define in those case
import time
from random import randint

def func(param):
    time.sleep(randint(1, 2))
    print('param:', param)
    return param

poi =  Parallel(func)
poi.put(1)
poi.put(3)
poi.run_as_async()
poi.puts([5, 7, 9])
# output rand ->
# get param 1
# get param 5
# ...
# get param 7

# if await with `keep` flag
poi.await_at(keep=True) # -> will block because the `keep` flag set will block until self.stop()
                     # its exit with poi.stop() in another thread
# else if only use await
poi.await_at() # -> will block until tasks were consumed finish

# if want to block put, set `queue.maxsize`

poi.queue.maxsize = poi.size

poi =  Parallel(func)
poi.puts([1, 2, 3])
poi.run_as_async()
print('start')
poi.await_at()
print('end')

# output ->
# start
# param: 2
# param: 1
# param: 3
# end

poi =  Parallel(func)
poi.puts([1, 2, 3])
poi.run_as_await()
print('start')
poi.put(None)

# output ->
# param: 2
# param: 1
# param: 3
# start
# raise ActorExitException: This actor is terminated

poi =  Parallel(func)
poi.run_as_async()
print('start')
poi.puts([1, 2, 3])

# output raise ->
# ActorExitException: This actor is terminated

poi =  Parallel(func, keep=True)
poi.run_as_async()
print('start')
poi.puts([1, 2, 3])
poi.await_at()
print('end')

# output ->
# start
# param: 2
# param: 1
# param: 3  <- blocked in here until poi.stop()

poi =  Parallel(func, keep=True)
poi.run_as_async()
print('start')
poi.puts([1, 2, 3])
time.sleep(3)
print('sleep timeout')
poi.await_at(keep=False)
print('end')

# output ->
# start
# param: 2
# param: 1
# param: 3
# sleep timeout
# end

poi =  Parallel(func, preload=[1, 2, 3])
print('start')
poi.run_as_async()
time.sleep(3)
print('sleep timeout')
poi.await_at(keep=True)
print('end')

# output ->
# start
# param: 2
# param: 1
# param: 3
# sleep timeout
# end

poi =  Parallel(func, preload=[1, 2, 3])
print('start')
poi.run_as_async()
poi.await_at(keep=True)
print('end')

# output ->
# start
# param: 2
# param: 1
# param: 3 <- blocked in here until poi.stop()

poi =  Parallel(func, preload=[1, 2, 3])
...
# equal Parallel(func).puts([1, 2, 3])

poi =  Parallel(func, preload=[1, 2, 3])
poi.run_as_async()
results = list(poi.results)
print('results', results)
for item in poi.results:
    print('item', item)
print('tasks end')

# output ->
# results [2, 1, 3]
# tasks end

poi =  Parallel(func, preload=[1, 2, 3])
poi.run_as_async()
for item in poi.results:
    print('item', item)
print('tasks end')

# output ->
# item 2
# item 1
# item 3
# tasks end

poi =  Parallel(func, preload=range(20), progress=True)
poi.run_as_async()

# output ->
# 100%|███████████████████████| 20/20 [00:01<00:00,  20it/s]

daemon

make current process run as daemon

Note

need install python-daemon

https://www.python.org/dev/peps/pep-3143

https://pypi.org/project/python-daemon/

Usage:

# deamon.py
from sniputils.logsetting import reset_logbase
reset_logbase('xxx.log')

from sniputils.deamon import daemon
with daemon:
    main()

# use reset_logbase to set logging file
$ export DAEMON_ERR_LOG='error.log'
$ python deamon.py

# its like the command via nohup
$ nohup python -m main > xxx.log 2> error.log &

sniputils.daemon.daemon = <sniputils.daemon.Daemon object>

import_hook

sniputils.import_hook.which_import_me()

get which import the caller who used this track func

Usage:

# foo.py
from .bar import *

# bar.py
from import_hook import which_import_me
back_imported = which_import_me()
print(back_imported)

$ python bar  --> output: <module foo.py>

Returns:module - which import the caller

sniputils.import_hook.set_reimport(module)

set a module to re-importable

Parameters:module (str) – module name which want to reimportable

Usage:

from sniputils.import_hook import set_reimport
set_reimport(__name__)

sniputils.import_hook.disreimport(module)

disable a re-importable module

usage like set_reimport

Parameters:module (str) – module name which want to disable reimport

import_hook.upstream

append upstream folder path in env PATH while folder is module

to resolve relative import as absolute grammar

Note

inject monkey patch

support multiple to exec patch more times

Usage:

# as files like this:
/top
    __init__.py
    /foo
        __init__.py
        foo.py
    /bar
        __init__.py
        bar.py

# bar.py
import sniputils.import_hook.upstream
import foo.foo.xxx      # <- this is relative import but use absolute grammar

$ python bar
# it will append ['/top'] to sys.path

Note

NOTE: re-import this module will also reload and append to sys.path, it’s means

# in /test/foo/foo.py
import sniputils.import_hook.upstream
# in /retest/bar/bar.py
import sniputils.import_hook.upstream
# in /top.py
import test.foo.foo
import retest.bar.bar <- will append ['/test', '/retest'] to sys.path

snippets

sniputils.snippets.args2set(func)

class sniputils.snippets.classproperty

Bases: property

sniputils.snippets.ensure_dir_exist(file)

sniputils.snippets.ensure_set(data)

sniputils.snippets.except_all(item=None)

context to except all Exception

Parameters:item – any thing as return Returns:item

print('start')
with except_all():
    print(1 / 0)
    print('result')
print('end')

sniputils.snippets.period_split(start, end, delta: datetime.timedelta) → List[sniputils.snippets.period]

Parameters:

• start – period start time, string or datetime type
• end – period ending time, string or datetime type
• delta – steping length, must be a timedelta

Returns:

sniputils.snippets.reduce_set(items_list: List[list] = None) → set

flatten list of values list to set

Usage:

reduce_set([[1, 2, 3], [2, 3, 4], [5]])
# {1, 2, 3, 4, 5}

reduce_set([])
# set()

reduce_set()
# set()

Parameters:items_list (list) – list of values list Returns:flatten set