python.Decorators Namespace Reference

Functions
	_reraise_exception (new_exc, exc_info=None)
	forking (func)
Callable[[TFunc], TFunc]	deprecate (str reason, *, bool warn_once_per_call=True, bool print_context=False)

Variables
	log = logging.getLogger(__name__)
	TFunc = Callable[..., Any]

Detailed Description

Some useful decorators.

Function Documentation

_reraise_exception()

python.Decorators._reraise_exception ( new_exc, exc_info = None )

protected

Definition at line 17 of file Decorators.py.

def _reraise_exception(new_exc, exc_info=None):
    if exc_info is None:
        exc_info = sys.exc_info()
    _exc_class, _exc, tb = exc_info
    raise new_exc.__class__ (new_exc, tb)
    
 

deprecate()

Callable[[TFunc], TFunc] python.Decorators.deprecate	(	str	reason,
		*	,
		bool	warn_once_per_call = True,
		bool	print_context = False )

Decorator to indicate that a function should not be used

Parameters
----------
reason : str
    A string explaining why the function should not be used
warn_once_per_call : bool
    If True, only warn once per call location, default True
print_context: bool
    If True, print the calling context as part of the warning, default False

Definition at line 79 of file Decorators.py.

) -> Callable[[TFunc], TFunc]:
    """Decorator to indicate that a function should not be used
 
    Parameters
    ----------
    reason : str
        A string explaining why the function should not be used
    warn_once_per_call : bool
        If True, only warn once per call location, default True
    print_context: bool
        If True, print the calling context as part of the warning, default False
    """
 
    # Record when we've warned about a function+filename+line combination
    warncache: set[tuple[int, str, int]] = set()
 
    def deprecate_decorator(f):
        @functools.wraps(f)
        def call_deprecated(*args, **kwargs):
            """Call a deprecated function"""
            # Figure out where we're calling from. The stack indices are as follows
            # 0: this inspect.stack call
            # 1: the actual call site we want to mark
            frame_info = inspect.stack()[1]
            cache_value = id(f), frame_info.filename, frame_info.lineno
            if not warn_once_per_call or cache_value not in warncache:
                if warn_once_per_call:
                    warncache.add(cache_value)
                log.warning(f"Calling deprecated function '{f.__qualname__}'")
                log.warning(
                    f"in function {frame_info.function}, file {frame_info.filename}, line {frame_info.lineno}"
                )
                if print_context:
                    for idx, line in enumerate(frame_info.code_context):
                        log.warning(f"{frame_info.lineno + idx - frame_info.index}\t{line}")
                log.warning(reason)
            return f(*args, **kwargs)
        return call_deprecated
    return deprecate_decorator

forking()

python.Decorators.forking

(

func

)

This decorator implements the forking patterns, i.e. it runs the function
in a forked process.
see:
 http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/511474

Definition at line 24 of file Decorators.py.

def forking(func):
    """
    This decorator implements the forking patterns, i.e. it runs the function
    in a forked process.
    see:
     http://aspn.activestate.com/ASPN/Cookbook/Python/Recipe/511474
    """
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        import os
        import pickle
 
        # create a pipe which will be shared between parent and child
        pread, pwrite = os.pipe()
 
        # do fork
        pid = os.fork()
 
        
        if pid > 0:
            os.close(pwrite)
            with os.fdopen(pread, 'rb') as f:
                status, result = pickle.load(f)
            os.waitpid(pid, 0)
            if status == 0:
                return result
            else:
                remote_exc = result[0]
                _reraise_exception(remote_exc)
 
        
        else:
            os.close(pread)
            try:
                result = func(*args, **kwargs)
                status = 0
            except (Exception, KeyboardInterrupt) as exc:
                import traceback
                exc_string = traceback.format_exc(limit=10)
                for l in exc_string.splitlines():
                    print ("[%d]"%os.getpid(),l.rstrip())
                sys.stdout.flush()
                result = exc, exc_string
                status = 1
            with os.fdopen(pwrite, 'wb') as f:
                try:
                    pickle.dump((status,result), f, pickle.HIGHEST_PROTOCOL)
                except pickle.PicklingError as exc:
                    pickle.dump((2,exc), f, pickle.HIGHEST_PROTOCOL)
            os._exit(0)
        return result
 
    return wrapper
 
 

Variable Documentation

log

python.Decorators.log = logging.getLogger(__name__)

Definition at line 13 of file Decorators.py.

TFunc

python.Decorators.TFunc = Callable[..., Any]

Definition at line 15 of file Decorators.py.