Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def run(self, args, **kwargs):
schema = self.app.client.managers['ConfigSchema'].get_by_ref_or_id(args.name, **kwargs)
if not schema:
msg = '%s "%s" doesn\'t exist or doesn\'t have a config schema defined.'
raise resource.ResourceNotFoundError(
msg % (self.resource.get_display_name(), args.name)
)
config = interactive.InteractiveForm(schema.attributes).initiate_dialog()
message = '---\nDo you want to preview the config in an editor before saving?'
description = 'Secrets will be shown in plain text.'
preview_dialog = interactive.Question(message, {'default': 'y', 'description': description})
if preview_dialog.read() == 'y':
try:
contents = yaml.safe_dump(config, indent=4, default_flow_style=False)
modified = editor.edit(contents=contents)
config = yaml.safe_load(modified)
except editor.EditorError as e:
print(six.text_type(e))
def run(self, args, **kwargs):
resource_id = getattr(args, self.pk_argument_name, None)
scope = getattr(args, 'scope', DEFAULT_CUD_SCOPE)
kwargs['params'] = {}
kwargs['params']['scope'] = scope
kwargs['params']['user'] = args.user
instance = self.get_resource(resource_id, **kwargs)
if not instance:
raise resource.ResourceNotFoundError(
'KeyValuePair with id "%s" not found' % resource_id
)
instance.id = resource_id # TODO: refactor and get rid of id
self.manager.delete(instance, **kwargs)
def run_and_print(self, args, **kwargs):
trace = None
try:
trace = self.run(args, **kwargs)
except resource.ResourceNotFoundError:
self.print_not_found(args.id)
raise OperationFailureException('Trace %s not found.' % (args.id))
# First filter for causation chains
trace = self._filter_trace_components(trace=trace, args=args)
# next filter for display purposes
trace = self._apply_display_filters(trace=trace, args=args)
return self.print_trace_details(trace=trace, args=args)
print('')
if required:
required = self._sort_parameters(parameters=parameters,
names=required)
print('Required Parameters:')
[self._print_param(name, parameters.get(name))
for name in required]
if optional:
optional = self._sort_parameters(parameters=parameters,
names=optional)
print('Optional Parameters:')
[self._print_param(name, parameters.get(name))
for name in optional]
except resource.ResourceNotFoundError:
print(('Action "%s" is not found. ' % args.ref_or_id) +
'Use "st2 action list" to see the list of available actions.')
except Exception as e:
print('ERROR: Unable to print help for action "%s". %s' %
(args.ref_or_id, e))
else:
self.parser.print_help()
return True
return False
def run(self, action, parameters=None, user=None, delay=None, trace_id=None, trace_tag=None):
parameters = parameters or {}
action_resource = self.client.managers['Action'].get_by_ref_or_id(action)
if not action_resource:
raise ResourceNotFoundError('Action "%s" cannot be found.'
% action)
action_ref = '.'.join([action_resource.pack, action_resource.name])
execution = Execution()
execution.action = action_ref
execution.parameters = parameters
if user:
execution.user = user
if delay:
execution.delay = delay
if not trace_id and trace_tag:
execution.context = {'trace_context': {'trace_tag': trace_tag}}
def run(self, args, **kwargs):
existing_execution = self.manager.get_by_id(args.id, **kwargs)
if not existing_execution:
raise resource.ResourceNotFoundError('Action execution with id "%s" cannot be found.' %
(args.id))
action_mgr = self.app.client.managers['Action']
runner_mgr = self.app.client.managers['RunnerType']
action_exec_mgr = self.app.client.managers['Execution']
action_ref = existing_execution.action['ref']
action = action_mgr.get_by_ref_or_id(action_ref)
runner = runner_mgr.get_by_name(action.runner_type)
action_parameters = self._get_action_parameters_from_args(action=action, runner=runner,
args=args)
execution = action_exec_mgr.re_run(execution_id=args.id,
parameters=action_parameters,
tasks=args.tasks,
def run_and_print(self, args, **kwargs):
try:
instance = self.run(args, **kwargs)
if not instance:
raise resource.ResourceNotFoundError("No matching items found")
self.print_output(
instance,
table.PropertyValueTable,
attributes=['all'],
json=args.json,
yaml=args.yaml,
)
except resource.ResourceNotFoundError:
print("No matching items found")
except Exception as e:
message = six.text_type(e)
print('ERROR: %s' % (message))
raise OperationFailureException(message)
action_mgr = self.app.client.managers['Action']
action_exec_mgr = self.app.client.managers['Execution']
if args.help:
action_ref_or_id = getattr(args, 'ref_or_id', None)
action_exec_id = getattr(args, 'id', None)
if action_exec_id and not action_ref_or_id:
action_exec = action_exec_mgr.get_by_id(action_exec_id, **kwargs)
args.ref_or_id = action_exec.action
if action_ref_or_id:
try:
action = action_mgr.get_by_ref_or_id(args.ref_or_id, **kwargs)
if not action:
raise resource.ResourceNotFoundError('Action %s not found' % args.ref_or_id)
runner_mgr = self.app.client.managers['RunnerType']
runner = runner_mgr.get_by_name(action.runner_type, **kwargs)
parameters, required, optional, _ = self._get_params_types(runner,
action)
print('')
print(textwrap.fill(action.description))
print('')
if required:
required = self._sort_parameters(parameters=parameters,
names=required)
print('Required Parameters:')
[self._print_param(name, parameters.get(name))
for name in required]
if optional:
optional = self._sort_parameters(parameters=parameters,