Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self):
registry = pyramid.threadlocal.get_current_registry()
settings = registry.settings
self.api_url = settings['api_url']
self.session = requests.Session()
:param service:
Cornice service to extract information from.
:rtype: dict
:returns: Path definition.
"""
path_obj = {}
path = service.path
route_name = getattr(service, 'pyramid_route', None)
# handle services that don't create fresh routes,
# we still need the paths so we need to grab pyramid introspector to
# extract that information
if route_name:
# avoid failure if someone forgets to pass registry
registry = self.pyramid_registry or get_current_registry()
route_intr = registry.introspector.get('routes', route_name)
if route_intr:
path = route_intr['pattern']
else:
msg = 'Route `{}` is not found by ' \
'pyramid introspector'.format(route_name)
raise ValueError(msg)
# handle traverse and subpath as regular parameters
# docs.pylonsproject.org/projects/pyramid/en/latest/narr/hybrid.html
for subpath_marker in ('*subpath', '*traverse'):
path = path.replace(subpath_marker, '{subpath}')
# Extract path parameters
parameters = self.parameters.from_path(path)
if parameters:
def get(self, configname, default_config=None):
""" Retrieve config from configuration file.
:param configname: configuration key to look for
:type configname: str
:param default_config: default config value to use if config not found
:type default_config: str|int|list|unicode
:return: value of requested configuration key.
"""
registry = pyramid.threadlocal.get_current_registry()
settings = registry.settings
config = None
try:
config = settings.get(configname) or default_config
except TypeError:
# We might hit this if we call registry too soon when initializing
pass
return config
def _get_view(self, view_name): # pragma: no cover
"""This code is copied from pyramid.view.
We trust it and don't test.
Returns True if a view with name view_name is registered for context.
"""
provides = [IViewClassifier] + map_(
providedBy,
(self.request, self.context)
)
try:
reg = self.request.registry
except AttributeError:
reg = get_current_registry()
return reg.adapters.lookup(provides, IView, name=view_name)
def invoke_callbacks_after_creation(self, callbacks=None):
reg = get_current_registry()
# If any of these callbacks throws an exception, the database
# transaction fails and so the Discussion object will not
# be added to the database (Discussion is not created).
known_callbacks = reg.getUtilitiesFor(IDiscussionCreationCallback)
if callbacks is not None:
known_callbacks = {
k: v for (k, v) in known_callbacks.iteritems() if k in callbacks}
for name, callback in known_callbacks:
callback.discussionCreated(self)
def get_queryable_attributes_keys(cls, lang):
queryable_attributes = []
if hasattr(cls, '__queryable_attributes__'):
settings = get_current_registry().settings
available_langs = settings['available_languages'].replace(' ', '|')
for attr in cls.__queryable_attributes__:
fallback_match = get_fallback_lang_match(
cls.__queryable_attributes__,
lang,
attr,
available_langs
)
if fallback_match not in queryable_attributes:
queryable_attributes.append(fallback_match)
return queryable_attributes
def service_entry(self):
"""Implement this as a property to have the context when looking for
the value of the setting"""
if self._service_entry is None:
settings = get_current_registry().settings
self._service_entry = settings.get('tokenserver.service_entry')
return self._service_entry
def load(self, name, newobject, registry=None):
""" A replace method used by the code that loads an existing dump.
Events sent during this replace will have a true ``loading`` flag."""
if registry is None:
registry = get_current_registry()
if name in self:
self.remove(name, loading=True)
self.add(name, newobject, loading=True, registry=registry)
def make_token(self, request, data):
backend = get_current_registry().getUtility(INodeAssignment)
email = data['email']
service = request.matchdict['application']
node, username = backend.get_node(email, service)
if node is None:
node, username = backend.create_node(email, service)
extra = {'service_entry': node}
token, secret, __ = super(NodeTokenManager, self)\
.make_token(request, data)
return token, secret, extra