Source code for pyramid.threadlocal

import threading

from pyramid.registry import global_registry


class ThreadLocalManager(threading.local):
    def __init__(self, default=None):
        # http://code.google.com/p/google-app-engine-django/issues/detail?id=119
        # we *must* use a keyword argument for ``default`` here instead
        # of a positional argument to work around a bug in the
        # implementation of _threading_local.local in Python, which is
        # used by GAE instead of _thread.local
        self.stack = []
        self.default = default

    def push(self, info):
        self.stack.append(info)

    set = push  # b/c

    def pop(self):
        if self.stack:
            return self.stack.pop()

    def get(self):
        try:
            return self.stack[-1]
        except IndexError:
            return self.default()

    def clear(self):
        self.stack[:] = []


def defaults():
    return {'request': None, 'registry': global_registry}


manager = ThreadLocalManager(default=defaults)


[docs] def get_current_request(): """ Return the currently active request or ``None`` if no request is currently active. This function should be used *extremely sparingly*, usually only in unit testing code. It's almost always usually a mistake to use ``get_current_request`` outside a testing context because its usage makes it possible to write code that can be neither easily tested nor scripted. """ return manager.get()['request']
[docs] def get_current_registry( context=None, ): # context required by getSiteManager API """ Return the currently active :term:`application registry` or the global application registry if no request is currently active. This function should be used *extremely sparingly*, usually only in unit testing code. It's almost always usually a mistake to use ``get_current_registry`` outside a testing context because its usage makes it possible to write code that can be neither easily tested nor scripted. """ return manager.get()['registry']
class RequestContext: def __init__(self, request): self.request = request def begin(self): request = self.request registry = request.registry manager.push({'registry': registry, 'request': request}) return request def end(self): manager.pop() def __enter__(self): return self.begin() def __exit__(self, *args): self.end()