Source code for cup.decorators

#!/usr/bin/env python
# -*- coding: utf-8 -*
# Copyright: [CUP] - See LICENSE for details.
# Authors: Guannan Ma (@mythmgn),
"""
:description:
    decorators related module
"""
from __future__ import print_function

import os
import time
import platform
import threading
from functools import wraps
from datetime import datetime as datetime_in


import cup
from cup import log

__all__ = [
    'Singleton', 'TraceUsedTime', 'needlinux', 'needposix', 'needmac',
    'py_versioncheck'
]


[docs]class Singleton(object): # pylint: disable=R0903 """ Make your class singeton example:: from cup import decorators @decorators.Singleton class YourClass(object): def __init__(self): pass """ def __init__(self, cls): self.__instance = None self.__cls = cls self._lock = threading.Lock() def __call__(self, *args, **kwargs): self._lock.acquire() if self.__instance is None: self.__instance = self.__cls(*args, **kwargs) self._lock.release() return self.__instance
[docs]def py_versioncheck(function, version): """ :platform: any platform + any functions in python :param version: The python on the OS should be >= param version. *E.g. version=('2', '7', '0')* OS python version should >= 2.7.0 """ ind = 0 py_version = platform.python_version_tuple() for i in py_version: if int(version(ind)) < int(i): raise cup.err.DecoratorException( 'Python version check failed. You expect version >= %s,' 'but python-version on this machine:%s' % (version, py_version) ) ind += 1 return function
[docs]def needlinux(function): """ 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 """ if platform.system() != 'Linux': raise cup.err.DecoratorException( 'The system is not linux.' 'This functionality only supported in linux' ) return function
[docs]def needposix(function): """ only support posix :platform: Posix compatible example :: from cup import decorators @decorators.needposix def your_func(): pass """ if os.name != 'posix': raise cup.err.DecoratorException( 'The system is not posix-based' ) return function
[docs]def needmac(function): """ only support macOS :platform: macOS example :: from cup import decorators @decorators.needmac def your_func(): pass """ if platform.system() != 'Darwin': raise cup.err.DecoratorException( 'The system is not macOS.' 'This functionality only supported in macOS' ) return function
# pylint:disable=R0903
[docs]class TraceUsedTime(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) """ def __init__(self, b_print_stdout=False, enter_msg='', leave_msg=''): """ :param b_print_stdout: When b_print_stdout is True, CUP will print to both LOGFILE that passed to cup.log.init_comlog and stdout :param enter_msg: entrance msg before invoking the function :param leave_msg: exist msg after leaving the function If you never use cup.log.init_comlog, make sure b_print_stdout == True """ self._b_print_stdout = b_print_stdout self._enter_msg = enter_msg self._leave_msg = leave_msg def __call__(self, function): @wraps(function) def _wrapper_log(*args, **kwargs): now = time.time() if self._b_print_stdout: print('**enter func:{0},time:{1}, msg:{2}'.format( function, datetime_in.now(), self._enter_msg )) cup.log.info( '**enter func:%s, msg:%s' % (function, self._enter_msg) ) function(*args, **kwargs) then = time.time() used_time = then - now cup.log.info( '**leave func:%s, used_time:%f, msg:%s' % ( function, used_time, self._enter_msg ) ) if self._b_print_stdout: print( '**leave func:{0}, time:{1}, used_time:{2}, ' 'msg:{3}'.format( function, datetime_in.now(), used_time, self._leave_msg) ) return _wrapper_log
# Things below for unittest @TraceUsedTime(False) def _test_trace_time(): """test trace time""" print('now, {0}, {1}'.format(time.time(), datetime_in.now())) time.sleep(3) print('then, {0}, {1}'.format(time.time(), datetime_in.now())) @TraceUsedTime(True) def _test_trace_time_log(): print('now, {0}, {1}'.format(time.time(), datetime_in.now())) time.sleep(3) print('then, {0}, {1}'.format(time.time(), datetime_in.now())) def _test_trace_time_map(sleep_time): print('ready to work') time.sleep(sleep_time) def _test(): cup.log.init_comlog( 'test', cup.log.DEBUG, './test.log', cup.log.ROTATION, 102400000, False ) _test_trace_time() _test_trace_time_log() func = TraceUsedTime( b_print_stdout=False, enter_msg='event_id: 0x12345', leave_msg='event_id: 0x12345' )(_test_trace_time_map) func(sleep_time=5) if __name__ == '__main__': _test() # vi:set tw=0 ts=4 sw=4 nowrap fdm=indent