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)
|
|
|
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']
|
|
|
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(object):
|
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()
|
