Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@wamp.register(u"io.timbr.kernel.{}.comm_msg".format(_key))
def comm_msg(self, *args, **kwargs):
msg = kwargs.get('msg', {})
log.msg("[comm_msg] {}".format(pformat(json_clean(msg))))
return client.shell_channel.send(msg)
@wamp.register(None)
def get_process_monitor(self, details=None):
self.log.debug("{cls}.get_process_monitor", cls=self.__class__.__name__)
if self._pmonitor:
return self._pmonitor.poll()
else:
emsg = "Could not retrieve process statistics: required packages not installed"
raise ApplicationError("crossbar.error.feature_unavailable", emsg)
@wamp.register(None)
def get_worker(self, worker_id, include_stats=False, details=None):
"""
Return detailed information about worker.
:param worker_id: ID of worker to get information for.
:type worker_id: str
:param include_stats: If true, include worker run-time statistics.
:type include_stats: bool
"""
if worker_id not in self._workers:
emsg = "No worker with ID '{}'".format(worker_id)
raise ApplicationError('crossbar.error.no_such_worker', emsg)
now = datetime.utcnow()
worker = self._workers[worker_id]
@wamp.register(None)
def get_router_component(self, id, details=None):
"""
Get details about a router component
:param id: The ID of the component to get
:type id: str
:param details: Call details.
:type details: :class:`autobahn.wamp.types.CallDetails`
:returns: Details of component
:rtype: dict
"""
self.log.debug("{name}.get_router_component({id})", name=self.__class__.__name__, id=id)
if id in self.components:
return self.components[id].marshal()
@wamp.register(u'mdstudio.auth.endpoint.verify')
def verify_claims(self, signed_claims):
try:
claims = jwt_decode(signed_claims, self.jwt_key)
except DecodeError:
return {'error': 'Could not verify user'}
except ExpiredSignatureError:
return {'expired': 'Request token has expired'}
return {'claims': claims}
@wamp.register('wamp.session.get')
def session_get(self, session_id, details=None):
"""
Get details for given session.
:param session_id: The WAMP session ID to retrieve details for.
:type session_id: int
:returns: WAMP session details.
:rtype: dict or None
"""
self.log.debug('wamp.session.get(session_id={session_id}, details={details})',
session_id=session_id, details=details)
if session_id in self._router._session_id_to_session:
session = self._router._session_id_to_session[session_id]
if not is_restricted_session(session):
@wamp.register(None)
@inlineCallbacks
def stop_component(self, component_id, details=None):
"""
Stop a component currently running within this container.
:param component_id: The ID of the component to stop.
:type component_id: int
:param details: Caller details.
:type details: instance of :class:`autobahn.wamp.types.CallDetails`
:returns: Stop information.
:rtype: dict
"""
self.log.debug('{klass}.stop_component({component_id}, {details})', klass=self.__class__.__name__, component_id=component_id, details=details)
@wamp.register(u'liestudio.haddock_docking.get')
def retrieve_haddock_project(self, project_name, session=None):
"""
Retrieve the docking results as tar zipped archive
:param project_name: HADDOCK web server project name
:type project_name: :py:str
"""
@wamp.register(u'liestudio.structure.retrieve_rcsb_structure')
def fetch_rcsb_structure(self, session=None, **kwargs):
"""
Download a structure file from the RCSB database using a PDB ID
"""
# Retrieve the WAMP session information
session = WAMPTaskMetaData(metadata=session).dict()
# Load configuration and update
config = _schema_to_data(BIOPYTHON_SCHEMA)
config.update(kwargs)
# Validate the configuration
try:
jsonschema.validate(config, BIOPYTHON_SCHEMA)
except ValueError, e:
@wamp.register(u'liestudio.scriptrunner.shell')
def shell_script_runner(self, script, session=None):
"""
Submit a new calculation to the ATB server
:param script: Shell script to run
:type script: :py:str
"""
# Return dummy data
session['status'] = 'done'
session['result'] = 'shell data'
return session