/usr/lib/python3/dist-packages/pecan/hooks.py is in python3-pecan 1.2.1-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 | import types
import sys
from inspect import getmembers
import six
from webob.exc import HTTPFound
from .util import iscontroller, _cfg
__all__ = [
'PecanHook', 'TransactionHook', 'HookController',
'RequestViewerHook'
]
def walk_controller(root_class, controller, hooks, seen=None):
seen = seen or set()
if type(controller) not in vars(six.moves.builtins).values():
# Avoid recursion loops
try:
if controller in seen:
return
seen.add(controller)
except TypeError:
# If we discover an unhashable item (like a list), it's not
# something that we want to traverse because it's not the sort of
# thing we would add a hook to
return
for hook in getattr(controller, '__hooks__', []):
# Append hooks from controller class definition
hooks.add(hook)
for name, value in getmembers(controller):
if name == 'controller':
continue
if name.startswith('__') and name.endswith('__'):
continue
if iscontroller(value):
for hook in hooks:
value._pecan.setdefault('hooks', set()).add(hook)
elif hasattr(value, '__class__'):
# Skip non-exposed methods that are defined in parent classes;
# they're internal implementation details of that class, and
# not actual routable controllers, so we shouldn't bother
# assigning hooks to them.
if (
isinstance(value, types.MethodType) and
any(filter(lambda c: value.__func__ in c.__dict__.values(),
value.__self__.__class__.mro()[1:]))
):
continue
walk_controller(root_class, value, hooks, seen)
class HookControllerMeta(type):
'''
A base class for controllers that would like to specify hooks on
their controller methods. Simply create a list of hook objects
called ``__hooks__`` as a member of the controller's namespace.
'''
def __init__(cls, name, bases, dict_):
hooks = set(dict_.get('__hooks__', []))
for base in bases:
# Add hooks from parent class and mixins
for hook in getattr(base, '__hooks__', []):
hooks.add(hook)
walk_controller(cls, cls, hooks)
HookController = HookControllerMeta(
'HookController',
(object,),
{'__doc__': ("A base class for controllers that would like to specify "
"hooks on their controller methods. Simply create a list "
"of hook objects called ``__hooks__`` as a class attribute "
"of your controller.")}
)
class PecanHook(object):
'''
A base class for Pecan hooks. Inherit from this class to create your
own hooks. Set a priority on a hook by setting the ``priority``
attribute for the hook, which defaults to 100.
'''
priority = 100
def on_route(self, state):
'''
Override this method to create a hook that gets called upon
the start of routing.
:param state: The Pecan ``state`` object for the current request.
'''
return
def before(self, state):
'''
Override this method to create a hook that gets called after
routing, but before the request gets passed to your controller.
:param state: The Pecan ``state`` object for the current request.
'''
return
def after(self, state):
'''
Override this method to create a hook that gets called after
the request has been handled by the controller.
:param state: The Pecan ``state`` object for the current request.
'''
return
def on_error(self, state, e):
'''
Override this method to create a hook that gets called upon
an exception being raised in your controller.
:param state: The Pecan ``state`` object for the current request.
:param e: The ``Exception`` object that was raised.
'''
return
class TransactionHook(PecanHook):
'''
:param start: A callable that will bind to a writable database and
start a transaction.
:param start_ro: A callable that will bind to a readable database.
:param commit: A callable that will commit the active transaction.
:param rollback: A callable that will roll back the active
transaction.
:param clear: A callable that will clear your current context.
A basic framework hook for supporting wrapping requests in
transactions. By default, it will wrap all but ``GET`` and ``HEAD``
requests in a transaction. Override the ``is_transactional`` method
to define your own rules for what requests should be transactional.
'''
def __init__(self, start, start_ro, commit, rollback, clear):
self.start = start
self.start_ro = start_ro
self.commit = commit
self.rollback = rollback
self.clear = clear
def is_transactional(self, state):
'''
Decide if a request should be wrapped in a transaction, based
upon the state of the request. By default, wraps all but ``GET``
and ``HEAD`` requests in a transaction, along with respecting
the ``transactional`` decorator from :mod:pecan.decorators.
:param state: The Pecan state object for the current request.
'''
controller = getattr(state, 'controller', None)
if controller:
force_transactional = _cfg(controller).get('transactional', False)
else:
force_transactional = False
if state.request.method not in ('GET', 'HEAD') or force_transactional:
return True
return False
def on_route(self, state):
state.request.error = False
if self.is_transactional(state):
state.request.transactional = True
self.start()
else:
state.request.transactional = False
self.start_ro()
def before(self, state):
if self.is_transactional(state) \
and not getattr(state.request, 'transactional', False):
self.clear()
state.request.transactional = True
self.start()
def on_error(self, state, e):
#
# If we should ignore redirects,
# (e.g., shouldn't consider them rollback-worthy)
# don't set `state.request.error = True`.
#
trans_ignore_redirects = (
state.request.method not in ('GET', 'HEAD')
)
if state.controller is not None:
trans_ignore_redirects = (
_cfg(state.controller).get(
'transactional_ignore_redirects',
trans_ignore_redirects
)
)
if type(e) is HTTPFound and trans_ignore_redirects is True:
return
state.request.error = True
def after(self, state):
if getattr(state.request, 'transactional', False):
action_name = None
if state.request.error:
action_name = 'after_rollback'
self.rollback()
else:
action_name = 'after_commit'
self.commit()
#
# If a controller was routed to, find any
# after_* actions it may have registered, and perform
# them.
#
if action_name:
controller = getattr(state, 'controller', None)
if controller is not None:
actions = _cfg(controller).get(action_name, [])
for action in actions:
action()
self.clear()
class RequestViewerHook(PecanHook):
'''
:param config: A (optional) dictionary that can hold ``items`` and/or
``blacklist`` keys.
:param writer: The stream writer to use. Can redirect output to other
streams as long as the passed in stream has a
``write`` callable method.
:param terminal: Outputs to the chosen stream writer (usually
the terminal)
:param headers: Sets values to the X-HTTP headers
Returns some information about what is going on in a single request. It
accepts specific items to report on but uses a default list of items when
none are passed in. Based on the requested ``url``, items can also be
blacklisted.
Configuration is flexible, can be passed in (or not) and can contain
some or all the keys supported.
**items**
This key holds the items that this hook will display. When this key is
passed only the items in the list will be used. Valid items are *any*
item that the ``request`` object holds, by default it uses the
following:
* path
* status
* method
* controller
* params
* hooks
.. note::
This key should always use a ``list`` of items to use.
**blacklist**
This key holds items that will be blacklisted based on ``url``. If
there is a need to omit urls that start with `/javascript`, then this
key would look like::
'blacklist': ['/javascript']
As many blacklisting items as needed can be contained in the list. The hook
will verify that the url is not starting with items in this list to display
results, otherwise it will get omitted.
.. note::
This key should always use a ``list`` of items to use.
For more detailed documentation about this hook, please see
:ref:`requestviewerhook`
'''
available = ['path', 'status', 'method', 'controller', 'params', 'hooks']
def __init__(self, config=None, writer=sys.stdout, terminal=True,
headers=True):
if not config:
self.config = {'items': self.available}
else:
if config.__class__.__name__ == 'Config':
self.config = config.to_dict()
else:
self.config = config
self.writer = writer
self.items = self.config.get('items', self.available)
self.blacklist = self.config.get('blacklist', [])
self.terminal = terminal
self.headers = headers
def after(self, state):
# Default and/or custom response information
responses = {
'controller': lambda self, state: self.get_controller(state),
'method': lambda self, state: state.request.method,
'path': lambda self, state: state.request.path,
'params': lambda self, state: [
(p[0].encode('utf-8'), p[1].encode('utf-8'))
for p in state.request.params.items()
],
'status': lambda self, state: state.response.status,
'hooks': lambda self, state: self.format_hooks(state.app.hooks),
}
is_available = [
i for i in self.items
if i in self.available or hasattr(state.request, i)
]
terminal = []
headers = []
will_skip = [
i for i in self.blacklist
if state.request.path.startswith(i)
]
if will_skip:
return
for request_info in is_available:
try:
value = responses.get(request_info)
if not value:
value = getattr(state.request, request_info)
else:
value = value(self, state)
except Exception as e:
value = e
terminal.append('%-12s - %s\n' % (request_info, value))
headers.append((request_info, value))
if self.terminal:
self.writer.write(''.join(terminal))
self.writer.write('\n\n')
if self.headers:
for h in headers:
key = str(h[0])
value = str(h[1])
name = 'X-Pecan-%s' % key
state.response.headers[name] = value
def get_controller(self, state):
'''
Retrieves the actual controller name from the application
Specific to Pecan (not available in the request object)
'''
path = state.request.pecan['routing_path'].split('/')[1:]
return state.controller.__str__().split()[2]
def format_hooks(self, hooks):
'''
Tries to format the hook objects to be more readable
Specific to Pecan (not available in the request object)
'''
str_hooks = [str(i).split()[0].strip('<') for i in hooks]
return [i.split('.')[-1] for i in str_hooks if '.' in i]
|