Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_deserialize_valid(self, request_, context):
from pyramid.traversal import resource_path
from adhocracy_core.sheets.principal import IUserExtended
inst = self.make_one().bind(request=request_, context=context)
context['user'] = testing.DummyResource(__provides__=IUserExtended)
cstrut = {'recipient': resource_path(context['user']),
'title': 'title',
'text': 'text'}
assert inst.deserialize(cstrut) == {'recipient': context['user'],
'title': 'title',
'text': 'text'}
def run(app, collections=None):
root = app.root_factory(app)
if not collections:
collections = DEFAULT_COLLECTIONS
for collection_name in collections:
collection = root[collection_name]
collection_path = resource_path(collection)
updated = 0
for count, item in enumerate(itervalues(collection)):
path = resource_path(item)
keys_add, keys_remove = item.update_keys()
update = False
if keys_add or keys_remove:
logger.debug('Updated keys: %s' % path)
update = True
rels_add, rels_remove = item.update_rels()
if rels_add or rels_remove:
logger.debug('Updated links: %s' % path)
update = True
if update:
updated += 1
logger.info('Collection %s: Updated %d of %d.' %
def filtered_elements(self, depth=1, ifaces: Iterable=None,
valuefilters: dict=None) -> Iterable:
"""See interface for docstring."""
system_catalog = find_catalog(self, 'system')
path_index = system_catalog['path']
query = path_index.eq(resource_path(self), depth=depth,
include_origin=False)
if ifaces:
interface_index = system_catalog['interfaces']
query &= interface_index.all(ifaces)
if valuefilters:
adhocracy_catalog = find_catalog(self, 'adhocracy')
for name, value in valuefilters.items():
index = adhocracy_catalog[name]
query &= index.eq(value)
resultset = query.execute()
for result in resultset:
yield result
def get_sheet(self, context: object,
isheet: IInterface,
request: Request=None,
creating: ResourceMetadata=None) -> IResourceSheet:
"""Get sheet for `context` and set the 'context' attribute.
:raise adhocracy_core.exceptions.RuntimeConfigurationError:
if there is no `isheet` sheet registered for context.
"""
if not creating and not isheet.providedBy(context):
msg = 'Sheet {0} is not provided by resource {1}'\
.format(isheet.__identifier__, resource_path(context))
raise RuntimeConfigurationError(msg)
meta = self.sheets_meta[isheet]
sheet = self._create_sheet(meta, context, request=request)
sheet.context = context
sheet.request = request
sheet.registry = self.registry
sheet.creating = creating
return sheet
def replace(self, catalog, reindex=False, output=None, **reindex_kw):
catalog_path = resource_path(catalog)
to_reindex = set()
changed = False
for index_name, index_factory in self.index_factories.items():
if index_name in catalog:
verb = 'replacing'
else:
verb = 'adding'
output and output(
'%s: %s index named %r' % (catalog_path, verb, index_name),
)
index = index_factory(self.name, index_name)
def _validate_post_pool(node, references: list, post_pool: IPool):
post_pool_path = resource_path(post_pool)
for reference in references:
if reference.__parent__ is post_pool:
continue
msg = 'You can only add references inside {}'.format(post_pool_path)
raise colander.Invalid(node, msg)
def _validate_post_pool(node, references: list, post_pool: IPool):
post_pool_path = resource_path(post_pool)
for reference in references:
if reference.__parent__ is post_pool:
continue
msg = 'You can only add references inside {}'.format(post_pool_path)
raise colander.Invalid(node, msg)
request: IRequest,
value: dict) -> [IRate]:
from adhocracy_core.resources.principal import get_system_user_anonymous
catalogs = find_service(context, 'catalogs')
anonymous = get_system_user_anonymous(request)
query = search_query._replace(
references=(Reference(None, IRate, 'subject', anonymous),
Reference(None, IRate, 'object', value['object'])),
resolve=True,
)
rates = catalogs.search(query).elements
rates_deanonymized = []
authenticated_user = request.anonymized_user or request.user
for rate in rates:
anonymized_creator = get_anonymized_creator(rate)
if anonymized_creator == resource_path(authenticated_user):
rates_deanonymized.append(rate)
return rates_deanonymized
def _get_new_version_created_in_this_transaction(registry, resource):
if not hasattr(registry, '_transaction_changelog'):
return
path = resource_path(resource)
changelog_metadata = registry._transaction_changelog[path]
return changelog_metadata.followed_by