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
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
|
import threading
import zope.component
from zope.component import getGlobalSiteManager
from zope.component.interfaces import ComponentLookupError
from zope.component.interfaces import IComponentLookup
from zope.component.registry import Components
from zope.component import getSiteManager as original_getSiteManager
from zope.interface import implements
from repoze.bfg.interfaces import ISettings
from repoze.bfg.zcml import zcml_configure
class ThreadLocalRegistryManager(threading.local):
registry = getGlobalSiteManager()
def set(self, registry):
self.registry = registry
def get(self):
return self.registry
def clear(self):
self.registry = getGlobalSiteManager()
registry_manager = ThreadLocalRegistryManager()
def setRegistryManager(manager): # for unit tests
global registry_manager
old_registry_manager = registry_manager
registry_manager = manager
return old_registry_manager
def makeRegistry(filename, package, options=None, lock=threading.Lock()):
# We push our ZCML-defined configuration into an app-local
# component registry in order to allow more than one bfg app to
# live in the same process space without one unnecessarily
# stomping on the other's component registrations (although I
# suspect directives that have side effects are going to fail).
# The only way to do that currently is to override
# zope.component.getGlobalSiteManager for the duration of the ZCML
# includes. We acquire a lock in case another make_app runs in a
# different thread simultaneously, in a vain attempt to prevent
# mixing of registrations. There's not much we can do about
# non-make_app code that tries to use the global site manager API
# directly in a different thread while we hold the lock. Those
# registrations will end up in our application's registry.
lock.acquire()
try:
registry = Components(package.__name__)
registry_manager.set(registry)
original_getSiteManager.sethook(getSiteManager)
zope.component.getGlobalSiteManager = registry_manager.get
zcml_configure(filename, package=package)
if options is None:
options = {}
settings = Settings(options)
registry.registerUtility(settings, ISettings)
return registry
finally:
zope.component.getGlobalSiteManager = getGlobalSiteManager
lock.release()
registry_manager.clear()
class Settings(object):
implements(ISettings)
def __init__(self, options):
self.reload_templates = options.get('reload_templates', False)
def getSiteManager(context=None):
if context is None:
return registry_manager.get()
else:
try:
return IComponentLookup(context)
except TypeError, error:
raise ComponentLookupError(*error.args)
def asbool(s):
s = str(s).strip()
return s.lower() in ('t', 'true', 'y', 'yes', 'on', '1')
def get_options(kw):
reload_templates = asbool(kw.get('reload_templates'))
return {'reload_templates':reload_templates}
from zope.testing.cleanup import addCleanUp
try:
addCleanUp(original_getSiteManager.reset)
except AttributeError:
# zope.hookable not yet installed
pass
addCleanUp(registry_manager.clear)
|
