cup package

@author Guannan Ma

Subpackages

Submodules

cup.cache module

description:

decorators related module

exception cup.cache.CacheFull(msg)[source]

Bases: BaseCupException

CacheFull for cache.KvCache

class cup.cache.KVCache(name=None, maxsize=0, time_extension=None)[source]

Bases: object

Key-Value Cache object.

You can use function set/get to access KeyValue Cache.

When a k-v is hit by function get, the expire_sec will be expanded to 2 * (expire_sec)

INFINITE_TIME = 315360000000
TIME_EXTENSION = 300
clear()[source]

remove all kv cache inside.

get(key)[source]

Get your cache with key. If the cache is expired, it will return None. If the key does not exist, it will return None.

pop_n_expired(num=0)[source]
Parameters:

num – if num is 0, will get all expired key/values

Returns:

A dict. Return expired items. Return type is a dict

{
    'key' : (value, expire_time)
}

set(kvdict, expire_sec=None)[source]

set cache with kvdict

{
    'key1': 'value1',
    'key2': 'value2',
    ....
}
Parameters:

  • kvdict – kvdict is a dict that contains your cache.

  • expire_sec – if expire_sec is None, the cache will never expire.

Returns:

True if set cache successfully. False otherwise.

set_time_extension(time_extension)[source]

set time extension

size()[source]
Returns:

cached item size

cup.cache.KvCache

alias of KVCache

cup.const module

internal const class

exception cup.const.ConstError(msg='')

Bases: BaseCupException

const error

cup.decorators module

description:

decorators related module

class cup.decorators.Singleton(cls)[source]

Bases: object

Make your class singeton

example:

from cup import decorators

@decorators.Singleton
class YourClass(object):
    def __init__(self):
    pass
class cup.decorators.TraceUsedTime(b_print_stdout=False, enter_msg='', leave_msg='')[source]

Bases: object

Trace used time inside a function.

Will print to LOGFILE if you initialized logging with cup.log.init_comlog.

example:

import time

from cup import decorators

@decorators.TraceUsedTime(True)
def test():
    print('test')
    time.sleep(4)


# trace something with context. E.g. event_id
def _test_trace_time_map(sleep_time):
    print('ready to work')
    time.sleep(sleep_time)


traced_test_trace_time_map = decorators.TraceUsedTime(
    b_print_stdout=False,
    enter_msg='event_id: 0x12345',
    leave_msg='event_id: 0x12345'
)(_test_trace_time_map)
traced_test_trace_time_map(sleep_time=5)
cup.decorators.needlinux(function)[source]

make sure the func is only used on linux. Raise cup.err.DecoratorException otherwise.

Platform:

Linux

example

from cup import decorators
@decorators.needlinux
def your_func():
    pass
cup.decorators.needmac(function)[source]

only support macOS

Platform:

macOS

example

from cup import decorators
@decorators.needmac
def your_func():
    pass
cup.decorators.needposix(function)[source]

only support posix

Platform:

Posix compatible

example

from cup import decorators
@decorators.needposix
def your_func():
    pass
cup.decorators.py_versioncheck(function, version)[source]
Platform:

any platform + any functions in python

Parameters:

version – The python on the OS should be >= param version. E.g. version=(‘2’, ‘7’, ‘0’) OS python version should >= 2.7.0

cup.err module

description:

error related module

exception cup.err.AccessDenied(str_resouce)[source]

Bases: ResException

Access Denied

exception cup.err.AsyncMsgError(msg='')[source]

Bases: NetException

cup.net.async msg related Exception

exception cup.err.BaseCupException(msg)[source]

Bases: Exception

base cup Exception. All other cup Exceptions will inherit this.

exception cup.err.ConfigError(msg='')[source]

Bases: BaseCupException

exception cup.err.DecoratorException(msg)[source]

Bases: BaseCupException

exception cup.err.LockFileError(msg='')[source]

Bases: BaseCupException

exception cup.err.LoggerException(msg)[source]

Bases: BaseCupException

Exception for logging, especially for cup.log

exception cup.err.NetException(msg='')[source]

Bases: BaseCupException

Network releated Exception

exception cup.err.NoSuchProcess(pid, str_process_name)[source]

Bases: ResException

No such Process Exception

exception cup.err.NotImplementedYet(msg='')[source]

Bases: BaseCupException

Not implemented yet

exception cup.err.NotSupportedError(feature/functionality not implemented by code yet)[source]

Bases: BaseCupException

exception cup.err.ResException(msg)[source]

Bases: BaseCupException

Resource releated Exception

exception cup.err.ThreadTermException(msg='')[source]

Bases: BaseCupException

Thread termination error

cup.exfile module

description:

file related functions

class cup.exfile.LockFile(fpath, locktype=2)[source]

Bases: object

Lock file in order to prevent others from trying to lock it again

Code Example:

from cup import exfile

filelock = exfile.LockFile()
# xxxx do something
filelock.lock(blocking=True)
# xxxxx do something else
filelock.unlock()
filepath()[source]

return filepath

lock(blocking=True)[source]

lock the file

Parameters:

blocking – If blocking is True, will block there until cup gets the lock. True by default.

Returns:

return False if locking fails

Raises:

Exception – raise cup.err.LockFileError if blocking is False and the lock action failed

unlock()[source]

unlock the locked file

cup.exfile.mk_newnode(abspath, check_exsistence=False)[source]

make new file node for abspath

Parameters:

  • abspath – plz use absolute path. Not relative path

  • check_exsistence – if True, will check if the abspath existence ( raise IOError if abspath exists)

Raises:

Exception – IOError

cup.exfile.safe_delete(abspath, not_del_list)[source]

the function will safely delete file/object of abspath.

If the abspath in not_del_list, safe delete will raise ValueError

Parameters:

  • abspath – pass in absolute path

  • not_del_list – cannot del path list

Raises:

Exception – ValueError, if abspath is in exfile.CANNOT_DEL_PATHLIST or not_del_list IOError, if cup encounters any problem

cup.exfile.safe_rmtree(abspath, not_del_list=None)[source]

the function will safely remove files/dirs of abspath with not_del_list excluded

Parameters:

  • abspath – pass in absolute path

  • not_del_list – cannot del path list

Raises:

Exception – ValueError, if abspath is in exfile.CANNOT_DEL_PATHLIST or not_del_list IOError, if cup encounters any problem

cup.flag module

description:

TypeMan and FlagMan is for someone who looks up value by key and the reverse (key by value)

class cup.flag.BaseMan[source]

Bases: object

for netmsg types

get_key_list()[source]

return key list

getkey_bynumber(number)[source]

get type by number

getnumber_bykey(key)[source]

get number by type

register_types(kvs)[source]

register types

class cup.flag.FlagMan[source]

Bases: BaseMan

msg flag class inherited from cup.flag.BaseMan inherited from cup.flag.BaseMan

class cup.flag.TypeMan[source]

Bases: BaseMan

msg flag class inherited from cup.flag.BaseMan

cup.log module

description:

common log related module

cup.log.backtrace_critical(msg, back_trace_len=0)[source]

logging.CRITICAL with backtrace support

cup.log.backtrace_debug(msg, back_trace_len=0)[source]

debug with backtrace support

cup.log.backtrace_error(msg, back_trace_len=0)[source]

error msg with backtarce support

cup.log.backtrace_info(msg, back_trace_len=0)[source]

info with backtrace support

cup.log.critical(msg, *args, **kwargs)[source]

Log a message with severity ‘CRITICAL’ on the root logger. If the logger has no handlers, call basicConfig() to add a console handler with a pre-defined format.

cup.log.critical_if(bol, msg, back_trace_len=1)[source]

log msg with critical loglevel if bol is true

cup.log.debug(msg, *args, **kwargs)[source]

Log a message with severity ‘DEBUG’ on the root logger. If the logger has no handlers, call basicConfig() to add a console handler with a pre-defined format.

cup.log.debug_if(bol, msg, back_trace_len=1)[source]

log msg with critical loglevel if bol is true

cup.log.error_if(bol, msg, back_trace_len=1)[source]

log msg with error loglevel if bol is true

cup.log.info(msg, *args, **kwargs)[source]

Log a message with severity ‘INFO’ on the root logger. If the logger has no handlers, call basicConfig() to add a console handler with a pre-defined format.

cup.log.info_if(bol, msg, back_trace_len=1)[source]

log msg with info loglevel if bol is true

cup.log.init_comlog(loggername, loglevel=20, logfile='cup.log', logtype=0, maxlogsize=1073741824, bprint_console=False, gen_wf=False)[source]

Initialize your default logger

Parameters:

  • loggername – Unique logger name for default logging.

  • loglevel – 4 default levels: log.DEBUG log.INFO log.ERROR log.CRITICAL

  • logfile – log file. Will try to create it if no existence

  • logtype –

    Two type candidiates: log.ROTATION and log.INFINITE

    log.ROTATION will let logfile switch to a new one (30 files at most). When logger reaches the 30th logfile, will overwrite from the oldest to the most recent.

    log.INFINITE will write on the logfile infinitely

  • maxlogsize – maxmum log size with byte

  • b_printcmd – print to stdout or not?

  • gen_wf – print log msges with level >= WARNING to file (${logfile}.wf)

E.g.

import logging
from cup import log
log.init_comlog(
    'test',
    log.DEBUG,
    '/home/work/test/test.log',
    log.ROTATION,
    1024,
    False
)
log.info('test xxx')
log.critical('test critical')
cup.log.parse(logline)[source]

return a dict if the line is valid. Otherwise, return None

Raises:

Exception – ValueError if logline is invalid

dict_info:= {
   'loglevel': 'DEBUG',
   'date': '2015-10-14',
   'time': '16:12:22,924',
   'pid': 8808,
   'tid': 1111111,
   'srcline': 'util.py:33',
   'msg': 'this is the log content',
   'tznum': 8,
   'tzstr': 'CST'
}
cup.log.reinit_comlog(loggername, loglevel=20, logfile='cup.log', logtype=0, maxlogsize=1073741824, bprint_console=False, gen_wf=False)[source]

reinitialize default root logger for cup logging

Parameters:

loggername – logger name, should be different from the original one

cup.log.setloglevel(logginglevel)[source]

change log level during runtime

from cup import log
log.setloglevel(log.DEBUG)
cup.log.warn_if(bol, msg, back_trace_len=1)[source]

log msg with error loglevel if bol is true

cup.log.warning(msg, *args, **kwargs)[source]

Log a message with severity ‘WARNING’ on the root logger. If the logger has no handlers, call basicConfig() to add a console handler with a pre-defined format.

cup.log.xcritical(loggername, msg, back_trace_len=1)[source]
Parameters:

  • loggername – shoule be xinit_comlog before calling xdebug/xinfo/xerror/xcritical

  • msg – log msg

Back_trace_len:

default 1, just ignore this param if you don’t know what it is. This param will trace back 1 layer and get the [code_filename:code_lines]

cup.log.xdebug(loggername, msg, back_trace_len=1)[source]
Parameters:

  • loggername – shoule be xinit_comlog before calling xdebug/xinfo/xerror/xcritical

  • msg – log msg

Back_trace_len:

1 by default, ignore this if you don’t need this

cup.log.xerror(loggername, msg, back_trace_len=1)[source]
Parameters:

  • loggername – shoule be xinit_comlog before calling xdebug/xinfo/xerror/xcritical

  • msg – log msg

Back_trace_len:

default 1, just ignore this param if you don’t know what it is. This param will trace back 1 layer and get the [code_filename:code_lines]

cup.log.xinfo(loggername, msg, back_trace_len=1)[source]
Parameters:

  • loggername – shoule be xinit_comlog before calling xdebug/xinfo/xerror/xcritical

  • msg – log msg

Back_trace_len:

default 1, just ignore this param if you don’t know what it is. This param will trace back 1 layer and get the [code_filename:code_lines]

cup.log.xinit_comlog(loggername, logger_params)[source]

xinit_comlog along with xdebug xinfo xwarn xerror are functions for different loggers other than logging.root (the global logger).

Parameters:

  • loggername – loggername, example: http.access,

  • logger_params – object of LoggerParams

Code example:

logparams = log.LoggerParams(
    log.DEBUG, 'cup.x.log', log.ROTATION, 100 * 1024 * 1024,
    True, True
)
log.xinit_comlog('log.x', logparams)
log.xdebug('log.x', 'xdebug')
log.xinfo('log.x', 'xinfo')
log.xerror('log.x', 'xerror')
logparams = log.LoggerParams(
    log.DEBUG, 'cup.y.log', log.ROTATION, 100 * 1024 * 1024,
    True, True
)
log.xinit_comlog('log.y', logparams)
log.xdebug('log.y', 'ydebug')
log.xinfo('log.y', 'yinfo')
log.xerror('log.y', 'yerror')
cup.log.xwarn(loggername, msg, back_trace_len=1)[source]
Parameters:

  • loggername – shoule be xinit_comlog before calling xdebug/xinfo/xerror/xcritical

  • msg – log msg

Back_trace_len:

default 1, just ignore this param if you don’t know what it is. This param will trace back 1 layer and get the [code_filename:code_lines]

cup.mail module

description:

mail related modules.

Recommand using SmtpMailer

class cup.mail.SmtpMailer(sender, server, port=25, is_html=False)[source]

Bases: object

Parameters:

  • sender – mail sender

  • server – smtp的mailserver

  • port – port

  • is_html – is html enabled

smtp server examples

from cup import mail
mailer = mail.SmtpMailer(
    'xxx@xxx.com',
    'xxxx.smtp.xxx.com',
    is_html=True
)
# if your smtp server needs login , pls uncomment line below:
# mailer.login(usernname, password)
mailer.sendmail(
    [
        'maguannan',
        'liuxuan05',
        'zhaominghao'
    ],
    'test_img',
    (
        'testset <img src="cid:screenshot.png"></img>'
    ),
    [
        '/home/work/screenshot.png',
        '../abc.zip'
    ]
)
login(username, passwords)[source]

if the smtp need login, plz call this method before you call sendmail

sendmail(recipients, subject='', body='', attachments=None, cc=None, bcc=None)[source]

send mail

Parameters:

  • recipients – “list” of recipients. See the example above

  • subject – subject

  • body – body of the mail

  • attachments – “list” of attachments. Plz use absolute file path!

  • cc – cc list

  • bcc – bcc list

Returns:

return (True, None) on success, return (False, error_msg) otherwise

setup(sender, server, port=25, is_html=False)[source]

change parameters during run-time

cup.mail.mutt_sendmail(tostr, subject, body, attach, content_is_html=False)[source]

Plz notice this function is not recommanded to use. Use SmtpMailer instead.

Parameters:

  • exec_cwd – exec working directory. Plz use

  • tostr – recipt list, separated by ,

  • subject – subject

  • body – email content

  • attach – email attachment

  • content_is_html – is htm mode opened

Returns:

return True on success, False otherwise

cup.platforms module

description:

cross-platform functions related module

cup.platforms.is_linux()[source]

Check if you are running on Linux.

Returns:

True or False

cup.platforms.is_mac()[source]

is mac os

cup.platforms.is_py2()[source]

is python 2.x

cup.platforms.is_py3()[source]

is python 3.x

cup.platforms.is_windows()[source]

Check if you are running on Windows.

Returns:

True or False

cup.thread module

description:

cup thread module

class cup.thread.CupThread(group=None, target=None, name=None, args=(), kwargs=None, *, daemon=None)[source]

Bases: Thread

CupThread is a sub-class inherited from threading.Thread;

Hint

CupThread has 3 more methods:

  1. raise_exc, to send a raise-exception signal to the thread, TRY to let the thread raise an exception.

  2. get_my_tid, get thread id

  3. terminate, to stop the thread

Caution

Notice if a thread is busy running under kernel-sysmode, it may not response to the signals! In other words, it may not raise any exception/terminate even though cup has send a CupThread signal!

get_my_tid()[source]

return thread id

raise_exc(exctype)[source]

asynchrously send ‘raise exception’ signal to the thread.

Parameters:

exctype – raise Exception, exctype type is class

Returns:

return 1 on success. 0 otherwise.

terminate(times=15)[source]

asynchrously terminate the thread.

Return True if the termination is successful or the thread is already stopped. Return False, otherwise.

Times:

retry times until call for failure.

class cup.thread.RWLock[source]

Bases: object

Read Write Lock is a typical lock type in computer world.

Code example:

from cup import thread
rwlock = thread.RWLock()
# can acquire read lock
rwlock.acquire_readlock()
# can acquire read lock again if there has not been WRITE locked
rwlock.acquire_readlock()   # <-- this will succeed
# rwlock.acquire_writelock()  # <--- this will hang if uncommented
rwlock.release_readlock()
rwlock.acquire_writelock()
# rwlock.acquire_readlock() # <--- this will hang if uncommented
rwlock.release_writelock()
acquire_readlock(wait_time=None)[source]

Acquire readlock.

Parameters:

wait_time – same to wait_time for acquire_writelock

Raise:

RuntimeError if after wait_time, cup still can NOT getthe lock

acquire_writelock(wait_time=None)[source]

Acquire write lock.

Important

If wait_time is not None and wait_time >=0, cup will wait until wait_time passes. If the call timeouts and cannot get the lock, will raise RuntimeError

release_readlock()[source]

release read lock

release_writelock()[source]

release write lock

cup.thread.async_raise(tid, exctype)[source]

Raises an exception in the threads with id tid

Parameters:

  • tid – thread id in python

  • exctype – exception class, e.g. IOError

cup.thread.thread_alive(obj)[source]

check if thread is alive. Py2 py3 compatible

Raises:

Exception – if the obejct does not have attr obj.is_alive and obj.isAlive, the lib will raise ValueError

cup.timeplus module

desc:

time related module. looking forward to accepting new patches

class cup.timeplus.TimePlus(timezone)[source]

Bases: object

TimePlus can convert printable time from 1 timezone 2 another including utc.

Code Example:

from __future__ import print_function
from cup import timeplus
import pytz
tz_shanghai = pytz.timezone('Aisa/Shanghai')
tplus = timeplus.TimePlus(tz_shanghai)
dtobj = datetime.datetime.now()
print(tplus.utc2local(dtobj))
get_timezone()[source]

return current pytz timezone object

local2utc(dateobj)[source]

local timezone to utc conversion

Returns:

a datetime.datetime object with utc timezone enabled

Raise:

ValueError if dateobj is not a datetime.datetime object

set_newtimezone(pytz_timezone)[source]

refresh timezone

Returns:

True if refreshing is done. False otherwise

utc2local(dateobj)[source]

utc datetime to local timezone datetime.datetime

classmethod utc_now()[source]

return utc_now

cup.timeplus.get_str_now(fmt='%Y-%m-%d-%H-%M-%S')[source]

return string of ‘now’

Parameters:

fmt – print-format, ‘%Y-%m-%d-%H-%M-%S’ by default

cup.unittest module

desc:

unittest module

class cup.unittest.CCaseExecutor[source]

Bases: object

Executor class for executing CUTCase test cases. See the example below

#!/usr/bin/env python

import sys
import os
import logging

import cup
import sb_global

from nfsinit import CClearMan
from nfs import CNfs
from cli import CCli

class TestMyCase(cup.unittest.CUTCase):
    def __init__(self):
        super(self.__class__, self).__init__(
            logfile=sb_global.G_CASE_RUN_LOG,
            b_logstd=False
        )
        cup.log.info( 'Start to run ' + str(__file__ ) )
        self._cli = CCli()
        self._nfs = CNfs()
        self._uniq_strman = cup.util.CGeneratorMan()
        self._clearman = CClearMan()

    def setup(self):
        str_uniq = self._uniq_strman.get_uniqname()
        self._workfolder = os.path.abspath(os.getcwd())                     + '/' + str_uniq
        self._work_remote_folder =                     self._nfs.translate_path_into_remote_under_rw_folder(
            str_uniq
            )
        os.mkdir(self._workfolder)
        self._clearman.addfolder(self._workfolder)

    def _check_filemd5(self, src, dst):
        ret = os.system('/usr/bin/diff --brief %s %s' % (src, dst) )
        cup.unittest.assert_eq( 0, ret )

    def test_run(self):
        #
        # @author: maguannan
        #
        inited_bigfile = sb_global.G_NFS_DISK_RD +                     sb_global.G_TEST_BIGFILE
        bigfile = self._workfolder +                     '/test_big_file_offsite_write_random_write'
        self.tmpfile = sb_global.G_TMP4USE +                     '/test_big_file_offsite_write_random_write'
        os.system( 'cp %s %s' % (inited_bigfile, bigfile) )
        os.system( 'cp %s %s' % (bigfile, self.tmpfile) )
        times = 50
        baseOffsite = 1000
        fp0 = open(bigfile, 'r+b')
        fp1 = open(self.tmpfile, 'rb+')
        for i in range( 0, times ):
            fp0.seek(baseOffsite)
            fp1.seek(baseOffsite)
            fp0.write( 'a' * 100 )
            fp1.write( 'a' * 100 )
            baseOffsite += 1000
        fp0.close()
        fp1.close()

        self._check_filemd5(bigfile, self.tmpfile)

    def teardown(self):
        if self.get_result():
            # if case succeeded, do sth
            os.unlink(self.tmpfile)
            self._clearman.teardown()
        else:
            # if case failed, do sth else.
            print 'failed'
        cup.log.info( 'End running ' + str(__file__ ) )

if __name__ == '__main__':
    cup.unittest.CCaseExecutor().runcase(TestMyCase())
classmethod runcase(case)[source]

run the case

class cup.unittest.CUTCase(logfile='./test.log', b_logstd=False, b_debug=False)[source]

Bases: object

CUTCase is compatible with nosetests. You can inherit this class and implement your own TestClass.

Notice class method [set_result] will set case status to True/False after executing the case. Then you can get case status in teardown through calling class method [get_result]

get_result()[source]

get case status during case teardown

set_result(b_result)[source]

set case running status

setup()[source]

set up

teardown()[source]
test_run()[source]

test_run function

cup.unittest.assert_eq(val, exp, errmsg='')[source]

if val != exp, aseert False and print errmsg

cup.unittest.assert_eq_one(val, array, errmsg='')[source]

assert val equals one of the items in the [array]

cup.unittest.assert_false(val, errmsg='')[source]

val should be False. Assert False otherwise.

cup.unittest.assert_ge(val, exp, errmsg='')[source]

expect val >= exp

cup.unittest.assert_gt(val, exp, errmsg='')[source]

assert_gt, expect val > exp

cup.unittest.assert_le(val, exp, errmsg='')[source]

expect val <= exp

cup.unittest.assert_lt(val, exp, errmsg='')[source]

assert_lt, expect val < exp

cup.unittest.assert_ne(val, exp, errmsg='')[source]

expect val != exp

cup.unittest.assert_not_eq(val, exp, errmsg='')[source]

assert not equal to

cup.unittest.assert_true(val, errmsg='')[source]

If val is not True, assert False and print to stdout.

Plz notice, log.critical(errmsg) will be invoked if logging system has been initialized with cup.log.init_comlog.

cup.unittest.expect_raise(function, exception, *argc, **kwargs)[source]

expect raise exception

cup.version module

desc:

CUP version for Programming Use Only.