Source code for cup.util.context

#!/usr/bin/env python
# -*- coding: utf-8 -*
# Copyright: [CUP] - See LICENSE for details.
# Authors: Guannan Ma (@mythmgn),
"""
:description:
    context for threadpool
"""

from __future__ import division
import threading


__all__ = [
    'ContextManager', 'ContextTracker4Thread'
]


[docs]class ContextManager(object): """ context for function call stack """ def __init__(self): self.contexts = [{}]
[docs] def call_with_context(self, new_context, func, *args, **kwargs): """ context is a {} """ self.contexts.append(new_context) try: return func(*args, **kwargs) finally: self.contexts.pop()
[docs] def get_context(self, key): """ get the context that has the key """ for context in reversed(self.contexts): if key in context: return context[key] return None
[docs]class ContextTracker4Thread(object): """ thread switch tracker """ def __init__(self): self.local_res = threading.local()
[docs] def current_context(self): """ get current context """ try: return self.local_res.current_context except AttributeError: current = self.local_res.current_context = ContextManager() return current
[docs] def call_with_context(self, context, func, *args, **kwargs): """ call [func] and set up a context with it """ return self.current_context().call_with_context( context, func, *args, **kwargs )
[docs] def get_context(self, key): """ get the context by key """ return self.current_context().get_context(key)
def __repr__(self): tmpstr = '' stackind = 0 contexts = self.local_res.current_context.contexts for i in reversed(contexts): tmpstr += 'Stack ind:%d' % stackind stackind += 1 for key in i.keys(): tmpstr += ' %s:%s' % (key, i[key]) tmpstr += '\n' return tmpstr
# vi:set tw=0 ts=4 sw=4 nowrap fdm=indent