本文整理汇总了Python中pecan.request方法的典型用法代码示例。如果您正苦于以下问题：Python pecan.request方法的具体用法？Python pecan.request怎么用？Python pecan.request使用的例子？那么恭喜您, 这里精选的方法代码示例或许可以为您提供帮助。您也可以进一步了解该方法所在模块pecan的用法示例。

在下文中一共展示了pecan.request方法的24个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。

示例1: post_query

​点赞 6

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def post_query(self, q=None):

if q is not None:

try:

query = query_parser.parseString(q)

except pyparsing.ParseException:

api.abort(501, {"cause": "Not implemented error",

"detail": "q",

"reason": "Query not implemented"})

resource_type = query[0]

api.enforce("create resource type", {"name": resource_type})

schema = pecan.request.indexer.get_resource_type_schema()

rt = schema.resource_type_from_dict(resource_type, {}, 'creating')

try:

pecan.request.indexer.create_resource_type(rt)

except indexer.ResourceTypeAlreadyExists:

pass

pecan.response.status = 204

开发者ID:gnocchixyz，项目名称:gnocchi，代码行数:19，

示例2: get_measures_or_abort

​点赞 6

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def get_measures_or_abort(references, operations, start,

stop, granularity, needed_overlap, fill):

try:

return processor.get_measures(

pecan.request.storage,

references,

operations,

start, stop,

granularity, needed_overlap, fill)

except exceptions.UnAggregableTimeseries as e:

api.abort(400, e)

# TODO(sileht): We currently got only one metric for these exceptions but

# we can improve processor to returns all missing metrics at once, so we

# returns a list for the future

except storage.MetricDoesNotExist as e:

api.abort(404, {"cause": "Unknown metrics",

"detail": [str(e.metric.id)]})

except storage.AggregationDoesNotExist as e:

api.abort(404, {"cause": "Metrics with unknown aggregation",

"detail": [(str(e.metric.id), e.method)]})

开发者ID:gnocchixyz，项目名称:gnocchi，代码行数:22，

示例3: deserialize

​点赞 6

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def deserialize(expected_content_types=None):

if expected_content_types is None:

expected_content_types = ("application/json", )

mime_type, options = werkzeug.http.parse_options_header(

pecan.request.headers.get('Content-Type'))

if mime_type not in expected_content_types:

abort(415)

try:

params = json.load(pecan.request.body_file)

except Exception as e:

details = rest_exceptions.UnableToDecodeBody(e,

pecan.request.body_file)

LOG.warning(details.jsonify())

abort(400, details)

return params

开发者ID:gnocchixyz，项目名称:gnocchi，代码行数:18，

示例4: get_pagination_options

​点赞 6

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def get_pagination_options(params, default):

try:

opts = voluptuous.Schema({

voluptuous.Required(

"limit", default=pecan.request.conf.api.max_limit):

voluptuous.All(voluptuous.Coerce(int),

voluptuous.Range(min=1),

voluptuous.Clamp(

min=1, max=pecan.request.conf.api.max_limit)),

"marker": six.text_type,

voluptuous.Required("sort", default=default):

voluptuous.All(

voluptuous.Coerce(arg_to_list),

[six.text_type]),

}, extra=voluptuous.REMOVE_EXTRA)(params)

except voluptuous.Invalid as e:

abort(400, {"cause": "Argument value error",

"reason": str(e)})

opts['sorts'] = opts['sort']

del opts['sort']

return opts

开发者ID:gnocchixyz，项目名称:gnocchi，代码行数:23，

示例5: post

​点赞 6

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def post(self):

enforce("create archive policy rule", {})

ArchivePolicyRuleSchema = voluptuous.Schema({

voluptuous.Required("name"): six.text_type,

voluptuous.Required("metric_pattern"): six.text_type,

voluptuous.Required("archive_policy_name"): six.text_type,

})

body = deserialize_and_validate(ArchivePolicyRuleSchema)

enforce("create archive policy rule", body)

try:

ap = pecan.request.indexer.create_archive_policy_rule(

body['name'], body['metric_pattern'],

body['archive_policy_name']

)

except indexer.ArchivePolicyRuleAlreadyExists as e:

abort(409, six.text_type(e))

except indexer.NoSuchArchivePolicy as e:

abort(400, e)

location = "/archive_policy_rule/" + ap.name

set_resp_location_hdr(location)

pecan.response.status = 201

return ap

开发者ID:gnocchixyz，项目名称:gnocchi，代码行数:26，

示例6: _lookup

​点赞 6

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def _lookup(self, name, *remainder):

m = pecan.request.indexer.list_metrics(

details=True,

attribute_filter={"and": [

{"=": {"name": name}},

{"=": {"resource_id": self.resource_id}},

]})

if m:

return MetricController(m[0]), remainder

resource = pecan.request.indexer.get_resource(self.resource_type,

self.resource_id)

if resource:

abort(404, six.text_type(indexer.NoSuchMetric(name)))

else:

abort(404, six.text_type(indexer.NoSuchResource(self.resource_id)))

开发者ID:gnocchixyz，项目名称:gnocchi，代码行数:18，

示例7: etag_precondition_check

​点赞 6

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def etag_precondition_check(obj):

etag, lastmodified = obj.etag, obj.lastmodified

# NOTE(sileht): Checks and order come from rfc7232

# in webob, the '*' and the absent of the header is handled by

# if_match.__contains__() and if_none_match.__contains__()

# and are identique...

if etag not in pecan.request.if_match:

abort(412)

elif (not pecan.request.environ.get("HTTP_IF_MATCH")

and pecan.request.if_unmodified_since

and pecan.request.if_unmodified_since < lastmodified):

abort(412)

if etag in pecan.request.if_none_match:

if pecan.request.method in ['GET', 'HEAD']:

abort(304)

else:

abort(412)

elif (not pecan.request.environ.get("HTTP_IF_NONE_MATCH")

and pecan.request.if_modified_since

and (pecan.request.if_modified_since >=

lastmodified)

and pecan.request.method in ['GET', 'HEAD']):

abort(304)

开发者ID:gnocchixyz，项目名称:gnocchi，代码行数:26，

示例8: index

​点赞 6

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def index():

return {

"build": gnocchi.__version__,

"versions": [

{

"status": "CURRENT",

"links": [

{

"rel": "self",

"href": pecan.request.application_url + "/v1/"

}

],

"id": "v1.0",

"updated": "2015-03-19"

}

]

}

开发者ID:gnocchixyz，项目名称:gnocchi，代码行数:19，

示例9: get_resource

​点赞 6

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def get_resource(resource, resource_ident):

"""Get the resource from the uuid or logical name.

:param resource: the resource type.

:param resource_ident: the UUID or logical name of the resource.

:returns: The resource.

"""

resource = getattr(objects, resource)

context = pecan.request.context

if context.is_admin:

context.all_projects = True

if uuidutils.is_uuid_like(resource_ident):

return resource.get_by_uuid(context, resource_ident)

return resource.get_by_name(context, resource_ident)

开发者ID:openstack，项目名称:zun，代码行数:18，

示例10: get_one

​点赞 6

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def get_one(self, container_ident, request_ident, **kwargs):

"""Retrieve information about the action."""

context = pecan.request.context

policy.enforce(context, "container:actions",

action="container:actions")

container = api_utils.get_resource('Container', container_ident)

action = objects.ContainerAction.get_by_request_id(

context, container.uuid, request_ident)

if action is None:

raise exception.ResourceNotFound(name="Action", id=request_ident)

action_id = action.id

action = actions_view.format_action(action)

show_traceback = False

if policy.enforce(context, "container:action:events",

do_raise=False, action="container:action:events"):

show_traceback = True

events_raw = objects.ContainerActionEvent.get_by_action(context,

action_id)

action['events'] = [actions_view.format_event(evt, show_traceback)

for evt in events_raw]

return action

开发者ID:openstack，项目名称:zun，代码行数:27，

示例11: add_security_group

​点赞 6

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def add_security_group(self, container_ident, **security_group):

"""Add security group to an existing container.

:param container_ident: UUID or Name of a container.

:param security_group: security_group to be added to container.

"""

container = api_utils.get_resource('Container', container_ident)

check_policy_on_container(

container.as_dict(), "container:add_security_group")

utils.validate_container_state(container, 'add_security_group')

# check if security group already presnt in container

context = pecan.request.context

compute_api = pecan.request.compute_api

security_group_id = self._check_security_group(context, security_group)

if security_group_id in container.security_groups:

msg = _("Security group %(id)s has been added to container.") % {

'id': security_group_id}

raise exception.InvalidValue(msg)

compute_api.add_security_group(context, container,

security_group_id)

pecan.response.status = 202

开发者ID:openstack，项目名称:zun，代码行数:25，

示例12: remove_security_group

​点赞 6

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def remove_security_group(self, container_ident, **security_group):

"""Remove security group from an existing container.

:param container_ident: UUID or Name of a container.

:param security_group: security_group to be removed from container.

"""

container = api_utils.get_resource('Container', container_ident)

check_policy_on_container(

container.as_dict(), "container:remove_security_group")

utils.validate_container_state(container, 'remove_security_group')

context = pecan.request.context

compute_api = pecan.request.compute_api

security_group_id = self._check_security_group(context, security_group)

if security_group_id not in container.security_groups:

msg = _("Security group %(id)s was not added to container.") % {

'id': security_group_id}

raise exception.InvalidValue(msg)

compute_api.remove_security_group(context, container,

security_group_id)

pecan.response.status = 202

开发者ID:openstack，项目名称:zun，代码行数:23，

示例13: rename

​点赞 6

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def rename(self, container_ident, name):

"""Rename an existing container.

:param container_ident: UUID or Name of a container.

:param name: a new name for this container.

"""

container = api_utils.get_resource('Container', container_ident)

check_policy_on_container(container.as_dict(), "container:rename")

if container.name == name:

raise exception.Conflict('The new name for the container is the '

'same as the old name.')

container.name = name

context = pecan.request.context

container.save(context)

return view.format_container(context, pecan.request.host_url,

container)

开发者ID:openstack，项目名称:zun，代码行数:18，

示例14: resize_container

​点赞 6

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def resize_container(self, container_ident, **kwargs):

"""Resize an existing container.

:param container_ident: UUID or name of a container.

:param kwargs: cpu/memory to be updated.

"""

container = api_utils.get_resource('Container', container_ident)

check_policy_on_container(container.as_dict(),

"container:resize_container")

utils.validate_container_state(container, 'resize_container')

if 'memory' in kwargs:

kwargs['memory'] = str(kwargs['memory'])

if 'cpu' in kwargs:

kwargs['cpu'] = float(kwargs['cpu'])

context = pecan.request.context

compute_api = pecan.request.compute_api

compute_api.resize_container(context, container, kwargs)

pecan.response.status = 202

return view.format_container(context, pecan.request.host_url,

container)

开发者ID:openstack，项目名称:zun，代码行数:22，

示例15: rebuild

​点赞 6

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def rebuild(self, container_ident, **kwargs):

"""Rebuild container.

:param container_ident: UUID or Name of a container.

"""

container = api_utils.get_resource('Container', container_ident)

check_policy_on_container(container.as_dict(), "container:rebuild")

utils.validate_container_state(container, 'rebuild')

if kwargs.get('image'):

container.image = kwargs.get('image')

if kwargs.get('image_driver'):

utils.validate_image_driver(kwargs.get('image_driver'))

container.image_driver = kwargs.get('image_driver')

LOG.debug('Calling compute.container_rebuild with %s',

container.uuid)

run = True if container.status == consts.RUNNING else False

context = pecan.request.context

container.status = consts.REBUILDING

container.save(context)

compute_api = pecan.request.compute_api

compute_api.container_rebuild(context, container, run)

pecan.response.status = 202

开发者ID:openstack，项目名称:zun，代码行数:24，

示例16: reboot

​点赞 6

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def reboot(self, container_ident, timeout=None, **kwargs):

"""Reboot container.

:param container_ident: UUID or Name of a container.

"""

container = api_utils.get_resource('Container', container_ident)

check_policy_on_container(container.as_dict(), "container:reboot")

utils.validate_container_state(container, 'reboot')

LOG.debug('Calling compute.container_reboot with %s',

container.uuid)

context = pecan.request.context

container.status = consts.RESTARTING

container.save(context)

compute_api = pecan.request.compute_api

compute_api.container_reboot(context, container, timeout)

pecan.response.status = 202

开发者ID:openstack，项目名称:zun，代码行数:18，

示例17: _write_get_lines

​点赞 5

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def _write_get_lines():

encoding = pecan.request.headers.get('Transfer-Encoding', "").lower()

if encoding == "chunked":

# TODO(sileht): Support reading chunk without uwsgi when

# pecan.request.environ['wsgi.input_terminated'] is set.

# https://github.com/unbit/uwsgi/issues/1428

if uwsgi is None:

api.abort(

501, {"cause": "Not implemented error",

"reason": "This server is not running with uwsgi"})

return encoding, uwsgi.chunked_read()

return None, pecan.request.body

开发者ID:gnocchixyz，项目名称:gnocchi，代码行数:14，

示例18: ResourceTypeSchema

​点赞 5

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def ResourceTypeSchema(resource_type):

try:

pecan.request.indexer.get_resource_type(resource_type)

except indexer.NoSuchResourceType as e:

api.abort(400, e)

return resource_type

开发者ID:gnocchixyz，项目名称:gnocchi，代码行数:8，

示例19: abort

​点赞 5

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def abort(status_code, detail=''):

"""Like pecan.abort, but make sure detail is a string."""

if status_code == 404 and not detail:

raise RuntimeError("http code 404 must have 'detail' set")

if isinstance(detail, voluptuous.Invalid):

detail = {

'cause': 'Invalid input',

'reason': six.text_type(detail),

'detail': [six.text_type(path) for path in detail.path],

}

elif isinstance(detail, Exception):

detail = detail.jsonify()

LOG.debug("Aborting request. Code [%s]. Details [%s]", status_code, detail)

return pecan.abort(status_code, detail)

开发者ID:gnocchixyz，项目名称:gnocchi，代码行数:17，

示例20: set_resp_location_hdr

​点赞 5

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def set_resp_location_hdr(location):

location = '%s%s' % (pecan.request.script_name, location)

# NOTE(sileht): according the pep-3333 the headers must be

# str in py2 and py3 even this is not the same thing in both

# version

# see: http://legacy.python.org/dev/peps/pep-3333/#unicode-issues

if six.PY2 and isinstance(location, six.text_type):

location = location.encode('utf-8')

location = urllib_parse.quote(location)

pecan.response.headers['Location'] = location

开发者ID:gnocchixyz，项目名称:gnocchi，代码行数:12，

示例21: set_resp_link_hdr

​点赞 5

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def set_resp_link_hdr(marker, *args):

# NOTE(sileht): This comes from rfc5988.

# Setting prev, last is too costly/complicated, so just set next for now.

options = {}

for arg in args:

options.update(arg)

if "sorts" in options:

options["sort"] = options["sorts"]

del options["sorts"]

options["marker"] = marker

# NOTE(sileht): To always have the same orders

options = sorted(options.items())

params = urllib_parse.urlencode(options, doseq=True)

pecan.response.headers.add("Link", '; rel="next"' %

(pecan.request.path_url, params))

开发者ID:gnocchixyz，项目名称:gnocchi，代码行数:17，

示例22: get

​点赞 5

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def get(self):

ap = pecan.request.indexer.get_archive_policy(self.archive_policy)

if ap:

enforce("get archive policy", ap)

return ap

abort(404, six.text_type(

indexer.NoSuchArchivePolicy(self.archive_policy)))

开发者ID:gnocchixyz，项目名称:gnocchi，代码行数:9，

示例23: delete

​点赞 5

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def delete(self):

# NOTE(jd) I don't think there's any point in fetching and passing the

# archive policy here, as the rule is probably checking the actual role

# of the user, not the content of the AP.

enforce("delete archive policy", {})

try:

pecan.request.indexer.delete_archive_policy(self.archive_policy)

except indexer.NoSuchArchivePolicy as e:

abort(404, six.text_type(e))

except indexer.ArchivePolicyInUse as e:

abort(400, six.text_type(e))

开发者ID:gnocchixyz，项目名称:gnocchi，代码行数:13，

示例24: get_all

​点赞 5

​

# 需要导入模块: import pecan [as 别名]

# 或者: from pecan import request [as 别名]

def get_all(self):

enforce("list archive policy", {})

return pecan.request.indexer.list_archive_policies()

开发者ID:gnocchixyz，项目名称:gnocchi，代码行数:5，

注：本文中的pecan.request方法示例整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。