Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@rbac(('owner', 'user'))
def list_inputs(self):
"""Return a list of names of input values."""
if self._input_names is None:
self._input_names = [k for k, v in self.items(iotype='in')]
return self._input_names[:]
@rbac('owner')
def pack_zipfile(self, patterns, filename):
"""
Create ZipFile of files matching `patterns` if `filename` is legal.
patterns: list
List of :mod:`glob`-style patterns.
filename: string
Name of ZipFile to create.
"""
timeout = 5 * 60
return self._conn.invoke('pack_zipfile', (patterns, filename),
timeout=timeout)
@rbac(('owner', 'user'))
def collect_metadata(self):
# store metadata (size, etc.) for all relevant vars
self._get_all_var_metadata(self._reduced_graph)
for comp in self.get_comps():
if has_interface(comp, IAssembly):
comp.collect_metadata()
@rbac(('owner', 'user'))
def is_differentiable(self):
"""Return True if analytical derivatives can be
computed for this Component.
"""
if self.force_fd:
return False
return hasattr(self, 'provideJ')
@rbac(('owner', 'user'))
def get_itername(self):
"""Return current 'iteration coordinates'."""
return self.itername
@rbac('*')
def max_servers(self, resource_desc):
"""
Return the maximum number of servers which could be deployed for
`resource_desc`. The value needn't be exact, but performance may
suffer if it overestimates. The value is used to limit the number
of concurrent evaluations.
resource_desc: dict
Description of required resources.
"""
retcode, info = self.check_compatibility(resource_desc)
if retcode != 0:
return (0, info)
elif 'min_cpus' in resource_desc:
return (self.n_cpus / resource_desc['min_cpus'], {})
else:
@rbac(('owner', 'user'))
def post_setup(self):
pass
@rbac('owner', proxy_types=[Container])
def load_model(self, egg_filename):
"""
Load model from egg and return top-level object if this server's
`allow_shell` attribute is True.
egg_filename: string
Filename of egg to be loaded.
"""
self._logger.debug('load_model %r', egg_filename)
if not self._allow_shell:
self._logger.error('attempt to load %r by %r', egg_filename,
get_credentials().user)
raise RuntimeError('shell access is not allowed by this server')
self._check_path(egg_filename, 'load_model')
if self.tlo:
self.tlo.pre_delete()
@rbac('owner')
def stat(self, path):
"""
Returns ``os.stat(path)`` if `path` is legal.
path: string
Path to file to interrogate.
"""
self._logger.debug('stat %r', path)
self._check_path(path, 'stat')
try:
return os.stat(path)
except Exception as exc:
self._logger.error('stat %r in %s failed %s',
path, os.getcwd(), exc)
raise
@rbac('owner', proxy_types=[RemoteFile])
def open(self):
""" Open file for reading. """
mode = 'rb' if self.binary else 'rU'
return RemoteFile(open(self.abspath(), mode))