本文整理汇总了Python中st2common.util.api.get_full_public_api_url函数的典型用法代码示例。如果您正苦于以下问题:Python get_full_public_api_url函数的具体用法?Python get_full_public_api_url怎么用?Python get_full_public_api_url使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了get_full_public_api_url函数的17个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: assertCommonSt2EnvVarsAvailableInEnv
def assertCommonSt2EnvVarsAvailableInEnv(self, env):
"""
Method which asserts that the common ST2 environment variables are present in the provided
environment.
"""
for var_name in COMMON_ACTION_ENV_VARIABLES:
self.assertTrue(var_name in env)
self.assertEqual(env['ST2_ACTION_API_URL'], get_full_public_api_url())
开发者ID:AlexeyDeyneko,项目名称:st2,代码行数:9,代码来源:base.py
示例2: mock_st2_ctx
def mock_st2_ctx(self):
st2_ctx = {
'st2': {
'api_url': api_utils.get_full_public_api_url(),
'action_execution_id': uuid.uuid4().hex,
'user': cfg.CONF.system_user.user
}
}
return st2_ctx
开发者ID:nzlosh,项目名称:st2,代码行数:10,代码来源:workflow_inspection.py
示例3: _get_api_client
def _get_api_client(self):
"""
Retrieve API client instance.
"""
if not self._client:
ttl = (24 * 60 * 60)
temporary_token = create_token(username=self._api_username, ttl=ttl)
api_url = get_full_public_api_url()
self._client = Client(api_url=api_url, token=temporary_token.token)
return self._client
开发者ID:AlexeyDeyneko,项目名称:st2,代码行数:11,代码来源:datastore.py
示例4: get_api_client
def get_api_client(self):
"""
Retrieve API client instance.
"""
if not self._client:
self._logger.debug('Creating new Client object.')
api_url = get_full_public_api_url()
client = Client(api_url=api_url, token=self._auth_token)
self._client = client
return self._client
开发者ID:StackStorm,项目名称:st2,代码行数:12,代码来源:datastore.py
示例5: mock_st2_context
def mock_st2_context(self, ac_ex_db, context=None):
st2_ctx = {
'st2': {
'api_url': api_util.get_full_public_api_url(),
'action_execution_id': str(ac_ex_db.id),
'user': 'stanley'
}
}
if context:
st2_ctx['parent'] = context
return st2_ctx
开发者ID:nzlosh,项目名称:st2,代码行数:13,代码来源:base.py
示例6: _get_datastore_access_env_vars
def _get_datastore_access_env_vars(self):
"""
Return environment variables so datastore access using client (from st2client)
is possible with actions. This is done to be compatible with sensors.
:rtype: ``dict``
"""
env_vars = {}
if self.auth_token:
env_vars[AUTH_TOKEN_ENV_VARIABLE_NAME] = self.auth_token.token
env_vars[API_URL_ENV_VARIABLE_NAME] = get_full_public_api_url()
return env_vars
开发者ID:lyandut,项目名称:st2,代码行数:13,代码来源:python_runner.py
示例7: _get_api_client
def _get_api_client(self):
"""
Retrieve API client instance.
"""
token_expire = self._token_expire <= get_datetime_utc_now()
if not self._client or token_expire:
self._logger.audit('Creating new Client object.')
ttl = (24 * 60 * 60)
self._token_expire = get_datetime_utc_now() + timedelta(seconds=ttl)
temporary_token = create_token(username=self._api_username, ttl=ttl)
api_url = get_full_public_api_url()
self._client = Client(api_url=api_url, token=temporary_token.token)
return self._client
开发者ID:LindsayHill,项目名称:st2,代码行数:15,代码来源:datastore.py
示例8: _construct_workflow_execution_options
def _construct_workflow_execution_options(self):
# This URL is used by Mistral to talk back to the API
api_url = get_mistral_api_url()
endpoint = api_url + '/actionexecutions'
# This URL is available in the context and can be used by the users inside a workflow,
# similar to "ST2_ACTION_API_URL" environment variable available to actions
public_api_url = get_full_public_api_url()
# Build context with additional information
parent_context = {
'execution_id': self.execution_id
}
if getattr(self.liveaction, 'context', None):
parent_context.update(self.liveaction.context)
st2_execution_context = {
'api_url': api_url,
'endpoint': endpoint,
'parent': parent_context,
'notify': {},
'skip_notify_tasks': self._skip_notify_tasks
}
# Include notification information
if self._notify:
notify_dict = NotificationsHelper.from_model(notify_model=self._notify)
st2_execution_context['notify'] = notify_dict
if self.auth_token:
st2_execution_context['auth_token'] = self.auth_token.token
options = {
'env': {
'st2_execution_id': self.execution_id,
'st2_liveaction_id': self.liveaction_id,
'st2_action_api_url': public_api_url,
'__actions': {
'st2.action': {
'st2_context': st2_execution_context
}
}
}
}
return options
开发者ID:Plexxi,项目名称:st2,代码行数:46,代码来源:mistral_v2.py
示例9: _construct_st2_context
def _construct_st2_context(self):
st2_ctx = {
'st2': {
'action_execution_id': str(self.execution.id),
'api_url': api_util.get_full_public_api_url(),
'user': self.execution.context.get('user', cfg.CONF.system_user.user),
'pack': self.execution.context.get('pack', None)
}
}
if self.execution.context.get('api_user'):
st2_ctx['st2']['api_user'] = self.execution.context.get('api_user')
if self.execution.context:
st2_ctx['parent'] = self.execution.context
return st2_ctx
开发者ID:nzlosh,项目名称:st2,代码行数:17,代码来源:orquesta_runner.py
示例10: _get_common_action_env_variables
def _get_common_action_env_variables(self):
"""
Retrieve common ST2_ACTION_ environment variables which will be available to the action.
Note: Environment variables are prefixed with ST2_ACTION_* so they don't clash with CLI
environment variables.
:rtype: ``dict``
"""
result = {}
result['ST2_ACTION_PACK_NAME'] = self.get_pack_name()
result['ST2_ACTION_EXECUTION_ID'] = str(self.execution_id)
result['ST2_ACTION_API_URL'] = get_full_public_api_url()
if self.auth_token:
result['ST2_ACTION_AUTH_TOKEN'] = self.auth_token.token
return result
开发者ID:jonico,项目名称:st2,代码行数:18,代码来源:__init__.py
示例11: test_common_st2_env_vars_are_available_to_the_action
def test_common_st2_env_vars_are_available_to_the_action(self):
models = self.fixtures_loader.load_models(fixtures_pack="generic", fixtures_dict={"actions": ["local.yaml"]})
action_db = models["actions"]["local.yaml"]
runner = self._get_runner(action_db, cmd="echo $ST2_ACTION_API_URL")
runner.pre_run()
status, result, _ = runner.run({})
runner.post_run(status, result)
self.assertEquals(status, action_constants.LIVEACTION_STATUS_SUCCEEDED)
self.assertEqual(result["stdout"].strip(), get_full_public_api_url())
runner = self._get_runner(action_db, cmd="echo $ST2_ACTION_AUTH_TOKEN")
runner.pre_run()
status, result, _ = runner.run({})
runner.post_run(status, result)
self.assertEquals(status, action_constants.LIVEACTION_STATUS_SUCCEEDED)
self.assertEqual(result["stdout"].strip(), "mock-token")
开发者ID:rlugojr,项目名称:st2,代码行数:19,代码来源:test_localrunner.py
示例12: test_common_st2_env_vars_are_available_to_the_action
def test_common_st2_env_vars_are_available_to_the_action(self):
models = self.fixtures_loader.load_models(
fixtures_pack='generic', fixtures_dict={'actions': ['local.yaml']})
action_db = models['actions']['local.yaml']
runner = self._get_runner(action_db, cmd='echo $ST2_ACTION_API_URL')
runner.pre_run()
status, result, _ = runner.run({})
runner.post_run(status, result)
self.assertEquals(status, action_constants.LIVEACTION_STATUS_SUCCEEDED)
self.assertEqual(result['stdout'].strip(), get_full_public_api_url())
runner = self._get_runner(action_db, cmd='echo $ST2_ACTION_AUTH_TOKEN')
runner.pre_run()
status, result, _ = runner.run({})
runner.post_run(status, result)
self.assertEquals(status, action_constants.LIVEACTION_STATUS_SUCCEEDED)
self.assertEqual(result['stdout'].strip(), 'mock-token')
开发者ID:LindsayHill,项目名称:st2,代码行数:20,代码来源:test_localrunner.py
示例13: test_get_full_public_api_url
def test_get_full_public_api_url(self):
values = [
'http://foo.bar.com',
'http://foo.bar.com/',
'http://foo.bar.com:8080',
'http://foo.bar.com:8080/',
'http://localhost:8080/',
]
expected = [
'http://foo.bar.com/' + DEFAULT_API_VERSION,
'http://foo.bar.com/' + DEFAULT_API_VERSION,
'http://foo.bar.com:8080/' + DEFAULT_API_VERSION,
'http://foo.bar.com:8080/' + DEFAULT_API_VERSION,
'http://localhost:8080/' + DEFAULT_API_VERSION,
]
for mock_value, expected_result in zip(values, expected):
cfg.CONF.auth.api_url = mock_value
actual = get_full_public_api_url()
self.assertEqual(actual, expected_result)
开发者ID:ipv1337,项目名称:st2,代码行数:20,代码来源:test_util_api.py
示例14: _construct_workflow_execution_options
def _construct_workflow_execution_options(self):
# This URL is used by Mistral to talk back to the API
api_url = get_mistral_api_url()
endpoint = api_url + "/actionexecutions"
# This URL is available in the context and can be used by the users inside a workflow,
# similar to "ST2_ACTION_API_URL" environment variable available to actions
public_api_url = get_full_public_api_url()
# Build context with additional information
parent_context = {"execution_id": self.execution_id}
if getattr(self.liveaction, "context", None):
parent_context.update(self.liveaction.context)
st2_execution_context = {
"endpoint": endpoint,
"parent": parent_context,
"notify": {},
"skip_notify_tasks": self._skip_notify_tasks,
}
# Include notification information
if self._notify:
notify_dict = NotificationsHelper.from_model(notify_model=self._notify)
st2_execution_context["notify"] = notify_dict
if self.auth_token:
st2_execution_context["auth_token"] = self.auth_token.token
options = {
"env": {
"st2_execution_id": self.execution_id,
"st2_liveaction_id": self.liveaction_id,
"st2_action_api_url": public_api_url,
"__actions": {"st2.action": {"st2_context": st2_execution_context}},
}
}
return options
开发者ID:enykeev,项目名称:st2,代码行数:39,代码来源:v2.py
示例15: run
def run(self, action_parameters):
# Test connection
self._client.workflows.list()
# Setup inputs for the workflow execution.
inputs = self.runner_parameters.get('context', dict())
inputs.update(action_parameters)
# This URL is used by Mistral to talk back to the API
api_url = get_mistral_api_url()
endpoint = api_url + '/actionexecutions'
# This URL is available in the context and can be used by the users inside a workflow,
# similar to "ST2_ACTION_API_URL" environment variable available to actions
public_api_url = get_full_public_api_url()
# Build context with additional information
parent_context = {
'execution_id': self.execution_id
}
if getattr(self.liveaction, 'context', None):
parent_context.update(self.liveaction.context)
st2_execution_context = {
'endpoint': endpoint,
'parent': parent_context,
'notify': {},
'skip_notify_tasks': self._skip_notify_tasks
}
# Include notification information
if self._notify:
notify_dict = NotificationsHelper.from_model(notify_model=self._notify)
st2_execution_context['notify'] = notify_dict
if self.auth_token:
st2_execution_context['auth_token'] = self.auth_token.token
options = {
'env': {
'st2_execution_id': self.execution_id,
'st2_liveaction_id': self.liveaction_id,
'st2_action_api_url': public_api_url,
'__actions': {
'st2.action': {
'st2_context': st2_execution_context
}
}
}
}
# Get workbook/workflow definition from file.
with open(self.entry_point, 'r') as def_file:
def_yaml = def_file.read()
def_dict = yaml.safe_load(def_yaml)
is_workbook = ('workflows' in def_dict)
if not is_workbook:
# Non-workbook definition containing multiple workflows is not supported.
if len([k for k, _ in six.iteritems(def_dict) if k != 'version']) != 1:
raise Exception('Workflow (not workbook) definition is detected. '
'Multiple workflows is not supported.')
action_ref = '%s.%s' % (self.action.pack, self.action.name)
self._check_name(action_ref, is_workbook, def_dict)
def_dict_xformed = utils.transform_definition(def_dict)
def_yaml_xformed = yaml.safe_dump(def_dict_xformed, default_flow_style=False)
# Save workbook/workflow definition.
if is_workbook:
self._save_workbook(action_ref, def_yaml_xformed)
default_workflow = self._find_default_workflow(def_dict_xformed)
execution = self._client.executions.create(default_workflow,
workflow_input=inputs,
**options)
else:
self._save_workflow(action_ref, def_yaml_xformed)
execution = self._client.executions.create(action_ref,
workflow_input=inputs,
**options)
status = LIVEACTION_STATUS_RUNNING
partial_results = {'tasks': []}
# pylint: disable=no-member
current_context = {
'execution_id': str(execution.id),
'workflow_name': execution.workflow_name
}
exec_context = self.context
exec_context = self._build_mistral_context(exec_context, current_context)
LOG.info('Mistral query context is %s' % exec_context)
return (status, partial_results, exec_context)
开发者ID:azamsheriff,项目名称:st2,代码行数:96,代码来源:v2.py
示例16: _construct_workflow_execution_options
def _construct_workflow_execution_options(self):
# This URL is used by Mistral to talk back to the API
api_url = get_mistral_api_url()
endpoint = api_url + '/actionexecutions'
# This URL is available in the context and can be used by the users inside a workflow,
# similar to "ST2_ACTION_API_URL" environment variable available to actions
public_api_url = get_full_public_api_url()
# Build context with additional information
parent_context = {
'execution_id': self.execution_id
}
if getattr(self.liveaction, 'context', None):
parent_context.update(self.liveaction.context)
# Convert jinja expressions in the params of Action Chain under the parent context
# into raw block. If there is any jinja expressions, Mistral will try to evaulate
# the expression. If there is a local context reference, the evaluation will fail
# because the local context reference is out of scope.
chain_ctx = parent_context.get('chain') or {}
for attr in ['params', 'parameters']:
chain_params_ctx = chain_ctx.get(attr) or {}
for k, v in six.iteritems(chain_params_ctx):
parent_context['chain'][attr][k] = jinja.convert_jinja_to_raw_block(v)
st2_execution_context = {
'api_url': api_url,
'endpoint': endpoint,
'parent': parent_context,
'notify': {},
'skip_notify_tasks': self._skip_notify_tasks
}
# Include notification information
if self._notify:
notify_dict = NotificationsHelper.from_model(notify_model=self._notify)
st2_execution_context['notify'] = notify_dict
if self.auth_token:
st2_execution_context['auth_token'] = self.auth_token.token
options = {
'env': {
'st2_execution_id': self.execution_id,
'st2_liveaction_id': self.liveaction_id,
'st2_action_api_url': public_api_url,
'__actions': {
'st2.action': {
'st2_context': st2_execution_context
}
}
}
}
if not self.is_polling_enabled():
options['notify'] = [{'type': 'st2'}]
# Only used on reverse type workflows
task_name = self.runner_parameters.get('task_name', None)
if task_name is not None:
options['task_name'] = task_name
return options
开发者ID:nzlosh,项目名称:st2,代码行数:67,代码来源:mistral_v2.py
示例17: _spawn_sensor_process
def _spawn_sensor_process(self, sensor):
"""
Spawn a new process for the provided sensor.
New process uses isolated Python binary from a virtual environment
belonging to the sensor pack.
"""
sensor_id = self._get_sensor_id(sensor=sensor)
virtualenv_path = get_sandbox_virtualenv_path(pack=sensor['pack'])
python_path = get_sandbox_python_binary_path(pack=sensor['pack'])
if virtualenv_path and not os.path.isdir(virtualenv_path):
format_values = {'pack': sensor['pack'], 'virtualenv_path': virtualenv_path}
msg = PACK_VIRTUALENV_DOESNT_EXIST % format_values
raise Exception(msg)
trigger_type_refs = sensor['trigger_types'] or []
trigger_type_refs = ','.join(trigger_type_refs)
parent_args = json.dumps(sys.argv[1:])
args = [
python_path,
WRAPPER_SCRIPT_PATH,
'--pack=%s' % (sensor['pack']),
'--file-path=%s' % (sensor['file_path']),
'--class-name=%s' % (sensor['class_name']),
'--trigger-type-refs=%s' % (trigger_type_refs),
'--parent-args=%s' % (parent_args)
]
if sensor['poll_interval']:
args.append('--poll-interval=%s' % (sensor['poll_interval']))
env = os.environ.copy()
env['PYTHONPATH'] = get_sandbox_python_path(inherit_from_parent=True,
inherit_parent_virtualenv=True)
# Include full api URL and API token specific to that sensor
ttl = (24 * 60 * 60)
temporary_token = create_token(username='sensors_container', ttl=ttl)
env[API_URL_ENV_VARIABLE_NAME] = get_full_public_api_url()
env[AUTH_TOKEN_ENV_VARIABLE_NAME] = temporary_token.token
# TODO 1: Purge temporary token when service stops or sensor process dies
# TODO 2: Store metadata (wrapper process id) with the token and delete
# tokens for old, dead processes on startup
cmd = ' '.join(args)
LOG.debug('Running sensor subprocess (cmd="%s")', cmd)
# TODO: Intercept stdout and stderr for aggregated logging purposes
try:
process = subprocess.Popen(args=args, stdin=None, stdout=None,
stderr=None, shell=False, env=env,
preexec_fn=on_parent_exit('SIGTERM'))
except Exception as e:
cmd = ' '.join(args)
message = ('Failed to spawn process for sensor %s ("%s"): %s' %
(sensor_id, cmd, str(e)))
raise Exception(message)
self._processes[sensor_id] = process
self._sensors[sensor_id] = sensor
self._sensor_start_times[sensor_id] = int(time.time())
self._dispatch_trigger_for_sensor_spawn(sensor=sensor, process=process, cmd=cmd)
return process
开发者ID:hejin,项目名称:st2,代码行数:69,代码来源:process_container.py
注:本文中的st2common.util.api.get_full_public_api_url函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论