本文整理汇总了Python中st2common.util.jsonify.json_encode函数的典型用法代码示例。如果您正苦于以下问题：Python json_encode函数的具体用法？Python json_encode怎么用？Python json_encode使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了json_encode函数的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: callfunction
def callfunction(*args, **kwargs):
try:
args = list(args)
types = list(argtypes)
more = [args.pop(0)]
if len(types):
argspec = inspect.getargspec(f)
names = argspec.args[1:]
for name in names:
try:
a = args.pop(0)
more.append(types.pop(0)(a))
except IndexError:
try:
kwargs[name] = types.pop(0)(kwargs[name])
except IndexError:
LOG.warning("Type definition for '%s' argument of '%s' "
"is missing.", name, f.__name__)
except KeyError:
pass
body_cls = opts.get('body')
if body_cls:
if pecan.request.body:
try:
obj = body_cls(**pecan.request.json)
except jsonschema.exceptions.ValidationError as e:
return _handle_error(http_client.BAD_REQUEST, e)
more.append(obj)
else:
more.append(None)
args = tuple(more) + tuple(args)
status_code = opts.get('status_code')
noop_codes = [http_client.NOT_IMPLEMENTED,
http_client.METHOD_NOT_ALLOWED,
http_client.FORBIDDEN]
if status_code and status_code in noop_codes:
pecan.response.status = status_code
return json_encode(None)
try:
result = f(*args, **kwargs)
if status_code:
pecan.response.status = status_code
if content_type == 'application/json':
return json_encode(result)
else:
return result
except exc.HTTPException as e:
return _handle_error(e.wsgi_response.status, e)
except Exception as e:
return _handle_error(http_client.INTERNAL_SERVER_ERROR, e)
开发者ID:nagyist，项目名称:StackStorm-st2，代码行数:59，代码来源:base.py
示例2: callfunction
def callfunction(*args, **kwargs):
args = list(args)
types = copy.copy(arg_types)
more = [args.pop(0)]
if types:
argspec = inspect.getargspec(f)
names = argspec.args[1:]
for name in names:
try:
a = args.pop(0)
more.append(types.pop(0)(a))
except IndexError:
try:
kwargs[name] = types.pop(0)(kwargs[name])
except IndexError:
LOG.warning("Type definition for '%s' argument of '%s' "
"is missing.", name, f.__name__)
except KeyError:
pass
if body_cls:
if pecan.request.body:
data = pecan.request.json
else:
data = {}
obj = body_cls(**data)
try:
obj.validate()
except jsonschema.ValidationError as e:
raise exc.HTTPBadRequest(detail=e.message,
comment=traceback.format_exc())
except Exception as e:
raise exc.HTTPInternalServerError(detail=e.message,
comment=traceback.format_exc())
more.append(obj)
args = tuple(more) + tuple(args)
noop_codes = [http_client.NOT_IMPLEMENTED,
http_client.METHOD_NOT_ALLOWED,
http_client.FORBIDDEN]
if status_code and status_code in noop_codes:
pecan.response.status = status_code
return json_encode(None)
result = f(*args, **kwargs)
if status_code:
pecan.response.status = status_code
if content_type == 'application/json':
return json_encode(result, indent=None)
else:
return result
开发者ID:Kailashkatheth1，项目名称:st2，代码行数:57，代码来源:base.py
示例3: on_error
def on_error(self, state, e):
error_msg = getattr(e, 'comment', str(e))
LOG.debug('API call failed: %s', error_msg)
LOG.debug(traceback.format_exc())
if hasattr(e, 'body') and isinstance(e.body, dict):
body = e.body
else:
body = {}
if isinstance(e, exc.HTTPException):
status_code = state.response.status
message = str(e)
else:
status_code = httplib.INTERNAL_SERVER_ERROR
message = 'Internal Server Error'
body['faultstring'] = message
response_body = json_encode(body)
headers = state.response.headers or {}
headers['Content-Type'] = 'application/json'
headers['Content-Length'] = str(len(response_body))
return webob.Response(response_body, status=status_code, headers=headers)
开发者ID:Kailashkatheth1，项目名称:st2，代码行数:27，代码来源:hooks.py
示例4: on_error
def on_error(self, state, exc):
request_path = state.request.path
if cfg.CONF.api.serve_webui_files and request_path.startswith('/webui'):
# We want to return regular error response for requests to /webui
return
status_code = state.response.status
if status_code == httplib.NOT_FOUND:
message = 'The resource could not be found'
elif status_code == httplib.INTERNAL_SERVER_ERROR:
message = 'Internal Server Error'
else:
message = str(exc)
response_body = json_encode({'faultstring': message})
headers = state.response.headers or {}
if headers.get('Content-Type', None) == 'application/json':
# Already a JSON response
return
headers['Content-Type'] = 'application/json'
headers['Content-Length'] = str(len(response_body))
return webob.Response(response_body, status=status_code,
headers=headers)
开发者ID:srenatus，项目名称:st2，代码行数:26，代码来源:hooks.py
示例5: on_error
def on_error(self, state, e):
request_path = state.request.path
if cfg.CONF.api.serve_webui_files and request_path.startswith('/webui'):
# We want to return regular error response for requests to /webui
return
error_msg = getattr(e, 'comment', str(e))
LOG.debug('API call failed: %s', error_msg)
if hasattr(e, 'body') and isinstance(e.body, dict):
body = e.body
else:
body = {}
if isinstance(e, exc.HTTPException):
status_code = state.response.status
message = str(e)
else:
status_code = httplib.INTERNAL_SERVER_ERROR
message = 'Internal Server Error'
body['faultstring'] = message
response_body = json_encode(body)
headers = state.response.headers or {}
headers['Content-Type'] = 'application/json'
headers['Content-Length'] = str(len(response_body))
return webob.Response(response_body, status=status_code, headers=headers)
开发者ID:BlazeMediaGroup，项目名称:st2，代码行数:31，代码来源:hooks.py
示例6: _abort_other_errors
def _abort_other_errors(self, environ, start_response):
body = json_encode({
'faultstring': 'Internal Server Error'
})
headers = [('Content-Type', 'application/json')]
start_response('500 INTERNAL SERVER ERROR', headers)
return [body]
开发者ID:joshgre，项目名称:st2，代码行数:8，代码来源:auth.py
示例7: _abort_unauthorized
def _abort_unauthorized(self, environ, start_response):
body = json_encode({
'faultstring': 'Unauthorized'
})
headers = [('Content-Type', 'application/json')]
start_response('401 UNAUTHORIZED', headers)
return [body]
开发者ID:joshgre，项目名称:st2，代码行数:8，代码来源:auth.py
示例8: _abort_other_errors
def _abort_other_errors():
body = json_encode({
'faultstring': 'Internal Server Error'
})
headers = {}
headers['Content-Type'] = 'application/json'
status = httplib.INTERNAL_SERVER_ERROR
return webob.Response(body=body, status=status, headers=headers)
开发者ID:Pulsant，项目名称:st2，代码行数:9，代码来源:hooks.py
示例9: _abort_unauthorized
def _abort_unauthorized():
body = json_encode({
'faultstring': 'Unauthorized'
})
headers = {}
headers['Content-Type'] = 'application/json'
status = httplib.UNAUTHORIZED
return webob.Response(body=body, status=status, headers=headers)
开发者ID:jonico，项目名称:st2，代码行数:9，代码来源:hooks.py
示例10: format_output_object
def format_output_object(output_db_or_api):
if isinstance(output_db_or_api, ActionExecutionOutputDB):
data = ActionExecutionOutputAPI.from_model(output_db_or_api)
elif isinstance(output_db_or_api, ActionExecutionOutputAPI):
data = output_db_or_api
else:
raise ValueError('Unsupported format: %s' % (type(output_db_or_api)))
event = 'st2.execution.output__create'
result = 'event: %s\ndata: %s\n\n' % (event, json_encode(data, indent=None))
return result
开发者ID:nzlosh，项目名称:st2，代码行数:11，代码来源:executions.py
示例11: format
def format(gen):
message = '''event: %s\ndata: %s\n\n'''
for pack in gen:
if not pack:
# Note: gunicorn wsgi handler expect bytes, not unicode
yield six.binary_type(b'\n')
else:
(event, body) = pack
# Note: gunicorn wsgi handler expect bytes, not unicode
yield six.binary_type((message % (event, json_encode(body,
indent=None))).encode('utf-8'))
开发者ID:lyandut，项目名称:st2，代码行数:12，代码来源:stream.py
示例12: format
def format(gen):
# Yield initial state so client would receive the headers the moment it connects to the stream
yield '\n'
message = '''event: %s\ndata: %s\n\n'''
for pack in gen:
if not pack:
yield '\n'
else:
(event, body) = pack
yield message % (event, json_encode(body, indent=None))
开发者ID:BlazeMediaGroup，项目名称:st2，代码行数:12，代码来源:stream.py
示例13: on_error
def on_error(self, state, e):
if hasattr(e, 'body') and isinstance(e.body, dict):
body = e.body
else:
body = {}
if isinstance(e, exc.HTTPException):
status_code = state.response.status
message = str(e)
elif isinstance(e, db_exceptions.StackStormDBObjectNotFoundError):
status_code = httplib.NOT_FOUND
message = str(e)
elif isinstance(e, db_exceptions.StackStormDBObjectConflictError):
status_code = httplib.CONFLICT
message = str(e)
body['conflict-id'] = e.conflict_id
elif isinstance(e, rbac_exceptions.AccessDeniedError):
status_code = httplib.FORBIDDEN
message = str(e)
elif isinstance(e, (ValueValidationException, ValueError)):
status_code = httplib.BAD_REQUEST
message = getattr(e, 'message', str(e))
else:
status_code = httplib.INTERNAL_SERVER_ERROR
message = 'Internal Server Error'
# Log the error
is_internal_server_error = status_code == httplib.INTERNAL_SERVER_ERROR
error_msg = getattr(e, 'comment', str(e))
extra = {
'exception_class': e.__class__.__name__,
'exception_message': str(e),
'exception_data': e.__dict__
}
if is_internal_server_error:
LOG.exception('API call failed: %s', error_msg, extra=extra)
LOG.exception(traceback.format_exc())
else:
LOG.debug('API call failed: %s', error_msg, extra=extra)
if is_debugging_enabled():
LOG.debug(traceback.format_exc())
body['faultstring'] = message
response_body = json_encode(body)
headers = state.response.headers or {}
headers['Content-Type'] = 'application/json'
headers['Content-Length'] = str(len(response_body))
return webob.Response(response_body, status=status_code, headers=headers)
开发者ID:Pulsant，项目名称:st2，代码行数:53，代码来源:hooks.py
示例14: format
def format(gen):
# Yield initial state so client would receive the headers the moment it connects to the stream
yield '\n'
message = '''event: %s\ndata: %s\n\n'''
for pack in gen:
if not pack:
# Note: gunicorn wsgi handler expect bytes, not unicode
yield six.binary_type('\n')
else:
(event, body) = pack
# Note: gunicorn wsgi handler expect bytes, not unicode
yield six.binary_type(message % (event, json_encode(body, indent=None)))
开发者ID:azamsheriff，项目名称:st2，代码行数:14，代码来源:stream.py
示例15: __init__
def __init__(self, body=None, status=None, headerlist=None, app_iter=None, content_type=None,
*args, **kwargs):
# Do some sanity checking, and turn json_body into an actual body
if app_iter is None and body is None and ('json_body' in kwargs or 'json' in kwargs):
if 'json_body' in kwargs:
json_body = kwargs.pop('json_body')
else:
json_body = kwargs.pop('json')
body = json_encode(json_body).encode('UTF-8')
if content_type is None:
content_type = 'application/json'
super(Response, self).__init__(body, status, headerlist, app_iter, content_type,
*args, **kwargs)
开发者ID:lyandut，项目名称:st2，代码行数:15，代码来源:router.py
示例16: _abort_unauthorized
def _abort_unauthorized():
return webob.Response(json_encode({
'faultstring': 'Unauthorized'
}), status=401)
开发者ID:Kailashkatheth1，项目名称:st2，代码行数:4，代码来源:hooks.py
示例17: _json_body__set
def _json_body__set(self, value):
self.body = json_encode(value).encode('UTF-8')
开发者ID:lyandut，项目名称:st2，代码行数:2，代码来源:router.py
示例18: convert
def convert(self, items_list):
if not isinstance(items_list, list):
raise ValueError('Items to be converted should be a list.')
json_doc = json_encode(items_list)
return json_doc
开发者ID:StackStorm，项目名称:st2，代码行数:5，代码来源:json_converter.py
示例19: callfunction
def callfunction(*args, **kwargs):
function_name = f.__name__
args = list(args)
more = [args.pop(0)]
def cast_value(value_type, value):
if value_type == bool:
def cast_func(value):
return value.lower() in ['1', 'true']
else:
cast_func = value_type
result = cast_func(value)
return result
if body_cls:
if pecan.request.body:
data = pecan.request.json
obj = body_cls(**data)
try:
obj = obj.validate()
except (jsonschema.ValidationError, ValueError) as e:
raise exc.HTTPBadRequest(detail=e.message,
comment=traceback.format_exc())
except Exception as e:
raise exc.HTTPInternalServerError(detail=e.message,
comment=traceback.format_exc())
# Set default pack if one is not provided for resource create
if function_name == 'post' and not hasattr(obj, 'pack'):
extra = {
'resource_api': obj,
'default_pack_name': DEFAULT_PACK_NAME
}
LOG.debug('Pack not provided in the body, setting a default pack name',
extra=extra)
setattr(obj, 'pack', DEFAULT_PACK_NAME)
else:
obj = None
more.append(obj)
if arg_types:
# Cast and transform arguments based on the provided arg_types specification
result_args, result_kwargs = get_controller_args_for_types(func=f,
arg_types=arg_types,
args=args,
kwargs=kwargs)
more = more + result_args
kwargs.update(result_kwargs)
args = tuple(more) + tuple(args)
noop_codes = [http_client.NOT_IMPLEMENTED,
http_client.METHOD_NOT_ALLOWED,
http_client.FORBIDDEN]
if status_code and status_code in noop_codes:
pecan.response.status = status_code
return json_encode(None)
try:
result = f(*args, **kwargs)
except TypeError as e:
e = get_exception_for_type_error(func=f, exc=e)
raise e
if status_code:
pecan.response.status = status_code
if content_type == 'application/json':
if is_debugging_enabled():
indent = 4
else:
indent = None
return json_encode(result, indent=indent)
else:
return result
开发者ID:maniacs-ops，项目名称:st2，代码行数:78，代码来源:base.py
示例20: __call__
def __call__(self, environ, start_response):
# The middleware intercepts and handles all the errors happening down the call stack by
# converting them to valid HTTP responses with semantically meaningful status codes and
# predefined response structure (`{"faultstring": "..."}`). The earlier in the call stack is
# going to be run, the less unhandled errors could slip to the wsgi layer. Keep in mind that
# the middleware doesn't receive the headers that has been set down the call stack which
# means that things like CorsMiddleware and RequestIDMiddleware should be highier up the
# call stack to also apply to error responses.
try:
try:
return self.app(environ, start_response)
except NotFoundException:
raise exc.HTTPNotFound()
except Exception as e:
status = getattr(e, 'code', exc.HTTPInternalServerError.code)
if hasattr(e, 'detail') and not getattr(e, 'comment'):
setattr(e, 'comment', getattr(e, 'detail'))
if hasattr(e, 'body') and isinstance(getattr(e, 'body', None), dict):
body = getattr(e, 'body', None)
else:
body = {}
if isinstance(e, exc.HTTPException):
status_code = status
message = six.text_type(e)
elif isinstance(e, db_exceptions.StackStormDBObjectNotFoundError):
status_code = exc.HTTPNotFound.code
message = six.text_type(e)
elif isinstance(e, db_exceptions.StackStormDBObjectConflictError):
status_code = exc.HTTPConflict.code
message = six.text_type(e)
body['conflict-id'] = getattr(e, 'conflict_id', None)
elif isinstance(e, rbac_exceptions.AccessDeniedError):
status_code = exc.HTTPForbidden.code
message = six.text_type(e)
elif isinstance(e, (ValueValidationException, ValueError, ValidationError)):
status_code = exc.HTTPBadRequest.code
message = getattr(e, 'message', six.text_type(e))
else:
status_code = exc.HTTPInternalServerError.code
message = 'Internal Server Error'
# Log the error
is_internal_server_error = status_code == exc.HTTPInternalServerError.code
error_msg = getattr(e, 'comment', six.text_type(e))
extra = {
'exception_class': e.__class__.__name__,
'exception_message': six.text_type(e),
'exception_data': e.__dict__
}
if is_internal_server_error:
LOG.exception('API call failed: %s', error_msg, extra=extra)
else:
LOG.debug('API call failed: %s', error_msg, extra=extra)
if is_debugging_enabled():
LOG.debug(traceback.format_exc())
body['faultstring'] = message
response_body = json_encode(body)
headers = {
'Content-Type': 'application/json',
'Content-Length': str(len(response_body))
}
resp = Response(response_body, status=status_code, headers=headers)
return resp(environ, start_response)
开发者ID:nzlosh，项目名称:st2，代码行数:72，代码来源:error_handling.py
注：本文中的st2common.util.jsonify.json_encode函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论