File: threadlocal.py

package info (click to toggle)
python-pyramid 1.6+dfsg-1
  • links: PTS, VCS
  • area: main
  • in suites: stretch
  • size: 9,112 kB
  • ctags: 8,169
  • sloc: python: 41,764; makefile: 111; sh: 17
file content (60 lines) | stat: -rw-r--r-- 2,026 bytes parent folder | download | duplicates (2)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
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']