-
Notifications
You must be signed in to change notification settings - Fork 15
/
hooks.py
150 lines (119 loc) · 4.26 KB
/
hooks.py
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
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
##############################################################################
#
# Copyright (c) 2001, 2002 Zope Foundation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.1 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""Hooks for getting and setting a site in the thread global namespace.
"""
__docformat__ = 'restructuredtext'
import contextlib
import threading
from zope.component._compat import ZOPE_SECURITY_NOT_AVAILABLE_EX
try:
from zope.security.proxy import removeSecurityProxy
except ZOPE_SECURITY_NOT_AVAILABLE_EX: # pragma: no cover
def removeSecurityProxy(x):
return x
from zope.component.globalregistry import getGlobalSiteManager
from zope.interface.interfaces import ComponentLookupError
from zope.interface.interfaces import IComponentLookup
class read_property(object):
"""Descriptor for property-like computed attributes.
Unlike the standard 'property', this descriptor allows assigning a
value to the instance, shadowing the property getter function.
"""
def __init__(self, func):
self.func = func
def __get__(self, inst, cls):
if inst is None:
return self
return self.func(inst)
class SiteInfo(threading.local):
site = None
sm = getGlobalSiteManager()
@read_property
def adapter_hook(self):
adapter_hook = self.sm.adapters.adapter_hook
self.adapter_hook = adapter_hook
return adapter_hook
siteinfo = SiteInfo()
def setSite(site=None):
if site is None:
sm = getGlobalSiteManager()
else:
# We remove the security proxy because there's no way for
# untrusted code to get at it without it being proxied again.
# We should really look look at this again though, especially
# once site managers do less. There's probably no good reason why
# they can't be proxied. Well, except maybe for performance.
site = removeSecurityProxy(site)
# The getSiteManager method is defined by IPossibleSite.
sm = site.getSiteManager()
siteinfo.site = site
siteinfo.sm = sm
try:
del siteinfo.adapter_hook
except AttributeError:
pass
def getSite():
return siteinfo.site
@contextlib.contextmanager
def site(site):
old_site = getSite()
setSite(site)
try:
yield
finally:
setSite(old_site)
def getSiteManager(context=None):
"""A special hook for getting the site manager.
Here we take the currently set site into account to find the appropriate
site manager.
"""
if context is None:
return siteinfo.sm
# We remove the security proxy because there's no way for
# untrusted code to get at it without it being proxied again.
# We should really look look at this again though, especially
# once site managers do less. There's probably no good reason why
# they can't be proxied. Well, except maybe for performance.
sm = IComponentLookup(
context, getGlobalSiteManager())
sm = removeSecurityProxy(sm)
return sm
def adapter_hook(interface, object, name='', default=None):
try:
return siteinfo.adapter_hook(interface, object, name, default)
except ComponentLookupError:
return default
def setHooks():
from zope.component import _api
_api.adapter_hook.sethook(adapter_hook)
_api.getSiteManager.sethook(getSiteManager)
def resetHooks():
# Reset hookable functions to original implementation.
from zope.component import _api
_api.adapter_hook.reset()
_api.getSiteManager.reset()
# be sure the old adapter hook isn't cached, since
# it is derived from the SiteManager
try:
del siteinfo.adapter_hook
except AttributeError:
pass
# Clear the site thread global
clearSite = setSite
try:
from zope.testing.cleanup import addCleanUp
except ImportError: #pragma NO COVER
pass
else:
addCleanUp(resetHooks)