Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _cache_key(imports, target_trees, base_tree):
tree_paths = tuple(
(target_trees[target], paths) for target, paths in imports.items())
return compute_key({
'base_tree': base_tree,
'tree_paths': tree_paths,
})
@asyncio.coroutine
def _get_base_tree(self, runtime):
override_path = runtime.get_override(self.name)
if override_path is not None and not runtime.no_overrides:
return self._get_override_tree(runtime, override_path)
key = compute_key({
'type': self.type,
'plugin_fields': self.plugin_fields,
'peru_file': self.peru_file,
})
# Use a lock to prevent the same module from being double fetched. The
# lock is taken on the cache key, not the module itself, so two
# different modules with identical fields will take the same lock and
# avoid double fetching.
cache_key_lock = runtime.cache_key_locks[key]
with (yield from cache_key_lock):
if key in runtime.cache.keyval:
return runtime.cache.keyval[key]
with runtime.tmp_dir() as tmp_dir:
yield from plugin_fetch(
runtime.get_plugin_context(), self.type,
self.plugin_fields, tmp_dir,
async def _get_base_tree(self, runtime):
override_path = runtime.get_override(self.name)
if override_path is not None:
# Marking overrides as used lets us print a warning when an
# override is unused.
runtime.mark_override_used(self.name)
override_tree = await self._get_override_tree(
runtime, override_path)
return override_tree
key = compute_key({
'type': self.type,
'plugin_fields': self.plugin_fields,
'peru_file': self.peru_file,
})
# Use a lock to prevent the same module from being double fetched. The
# lock is taken on the cache key, not the module itself, so two
# different modules with identical fields will take the same lock and
# avoid double fetching.
cache_key_lock = runtime.cache_key_locks[key]
async with cache_key_lock:
# Skip reading the cache if --no-cache is set. This is the only
# place in the code we check that flag. Deterministic operations
# like tree merging still get read from cache, because there's no
# reason to redo them.
if key in runtime.cache.keyval and not runtime.no_cache:
return runtime.cache.keyval[key]
def rule_key(self, rule):
data = {
"remote": self.remote.fields,
"rule": rule.fields,
}
return compute_key(data)