我们从Python开源项目中，提取了以下11个代码示例，用于说明如何使用falcon.HTTP_METHODS。
def on_options(self, req, resp, **kwargs): """ Returns allowed methods in the Allow HTTP header. Also returns a JSON Schema, if supported by current resource. :param req: Falcon request :type req: falcon.request.Request :param resp: Falcon response :type resp: falcon.response.Response """ allowed_methods = [] for method in falcon.HTTP_METHODS: try: responder = getattr(self, 'on_' + method.lower()) except AttributeError: # resource does not implement this method pass else: # Usually expect a method, but any callable will do if callable(responder): allowed_methods.append(method) resp.set_header('Allow', ', '.join(sorted(allowed_methods))) result = {'name': self.objects_class.__name__} if self.objects_class.__doc__: result['description'] = self.objects_class.__doc__.strip() try: result['schema'] = self.get_schema(self.objects_class) except NotImplementedError: pass self.render_response(result, req, resp)
def _get_resource_methods(self, resource): allowed_methods = [] for method in HTTP_METHODS: if ( hasattr(resource, 'on_' + method.lower()) or resource is None ): allowed_methods.append(method) return allowed_methods
def before_operation(func): def _wrap_class_method(cls, method_name): method = getattr(cls, method_name, None) if method: setattr(cls, method_name, _before_operation(method)) def _before_operation(func_): cls = None is_cls = False if isinstance(func_, type): is_cls = True cls = func_ elif isinstance(func_, MethodType): cls = func_.__self__ def do_before(req, resp, **params): func(req, resp, cls, params) func_(req, resp, **params) if is_cls: methods = set() for path in cls.__schema__.values(): for method_name, method in path.items(): if method_name.upper() in HTTP_METHODS: op_name = method.get('operationId') if op_name: methods.add(op_name) for method in methods: _wrap_class_method(cls, method) return cls return do_before return _before_operation
def get_route_and_params(self, req): path_nodes = deque(self._split_uri(req.path)) params = dict() private_method_name = _build_private_method_name(req.method) nodes_tree = self._nodes route = None while path_nodes: path_node = path_nodes.popleft() self._raise_private_method_error(path_node) match = self._match_uri_node_template_and_set_params(nodes_tree, path_node, params) if match is None: break elif path_nodes: nodes_tree = nodes_tree[match] elif private_method_name not in nodes_tree[match]: re_string = '__({})__'.format('|'.join(HTTP_METHODS)) methods = [key.replace('_', '') \ for key in nodes_tree[match].keys() if re.match(re_string, key)] raise HTTPMethodNotAllowed(methods) else: route = nodes_tree[match][private_method_name] return route, params
def create_http_method_map(resource): """Maps HTTP methods (e.g., 'GET', 'POST') to methods of a resource object. Args: resource: An object with *responder* methods, following the naming convention *on_\**, that correspond to each method the resource supports. For example, if a resource supports GET and POST, it should define ``on_get(self, req, resp)`` and ``on_post(self, req, resp)``. Returns: dict: A mapping of HTTP methods to responders. """ method_map = {} for method in HTTP_METHODS: try: responder = getattr(resource, 'on_' + method.lower()) except AttributeError: # resource does not implement this method pass else: # Usually expect a method, but any callable will do if callable(responder): method_map[method] = responder # Attach a resource for unsupported HTTP methods allowed_methods = sorted(list(method_map.keys())) if 'OPTIONS' not in method_map: # OPTIONS itself is intentionally excluded from the Allow header opt_responder = responders.create_default_options(allowed_methods) method_map['OPTIONS'] = opt_responder allowed_methods.append('OPTIONS') na_responder = responders.create_method_not_allowed(allowed_methods) for method in HTTP_METHODS: if method not in allowed_methods: method_map[method] = na_responder return method_map
def before(action): """Decorator to execute the given action function *before* the responder. Args: action (callable): A function of the form ``func(req, resp, resource, params)``, where `resource` is a reference to the resource class instance associated with the request, and `params` is a dict of URI Template field names, if any, that will be passed into the resource responder as kwargs. Note: Hooks may inject extra params as needed. For example:: def do_something(req, resp, resource, params): try: params['id'] = int(params['id']) except ValueError: raise falcon.HTTPBadRequest('Invalid ID', 'ID was not valid.') params['answer'] = 42 """ def _before(responder_or_resource): if isinstance(responder_or_resource, six.class_types): resource = responder_or_resource for method in HTTP_METHODS: responder_name = 'on_' + method.lower() try: responder = getattr(resource, responder_name) except AttributeError: # resource does not implement this method pass else: # Usually expect a method, but any callable will do if callable(responder): # This pattern is necessary to capture the current # value of responder in the do_before_all closure; # otherwise, they will capture the same responder # variable that is shared between iterations of the # for loop, above. def let(responder=responder): do_before_all = _wrap_with_before(action, responder) setattr(resource, responder_name, do_before_all) let() return resource else: responder = responder_or_resource do_before_one = _wrap_with_before(action, responder) return do_before_one return _before
def after(action): """Decorator to execute the given action function *after* the responder. Args: action (callable): A function of the form ``func(req, resp, resource)``, where `resource` is a reference to the resource class instance associated with the request """ def _after(responder_or_resource): if isinstance(responder_or_resource, six.class_types): resource = responder_or_resource for method in HTTP_METHODS: responder_name = 'on_' + method.lower() try: responder = getattr(resource, responder_name) except AttributeError: # resource does not implement this method pass else: # Usually expect a method, but any callable will do if callable(responder): def let(responder=responder): do_after_all = _wrap_with_after(action, responder) setattr(resource, responder_name, do_after_all) let() return resource else: responder = responder_or_resource do_after_one = _wrap_with_after(action, responder) return do_after_one return _after # ----------------------------------------------------------------------------- # Helpers # -----------------------------------------------------------------------------
def _set_routes(cls): SWAGGER_VALIDATOR.validate(cls.__schema__) cls.__routes__ = set() cls.__options_routes__ = set() dict_ = defaultdict(list) schema = cls.__schema__ cls._set_key() if not hasattr(cls, '__schema_dir__'): cls.__schema_dir__ = get_module_path(cls) for uri_template in schema: all_methods_parameters = schema[uri_template].get('parameters', []) for method_name in HTTP_METHODS: method_schema = schema[uri_template].get(method_name.lower()) if method_schema: method_schema = deepcopy(method_schema) operation_id = method_schema['operationId'] try: getattr(cls, operation_id) except AttributeError: raise ModelBaseError("'operationId' '{}' was not found".format(operation_id)) definitions = schema.get('definitions') parameters = method_schema.get('parameters', []) parameters.extend(all_methods_parameters) method_schema['parameters'] = parameters route = Route(uri_template, method_name, operation_id, cls, method_schema, definitions, cls.__authorizer__) cls.__routes__.add(route) routes = defaultdict(set) for route in cls.__routes__: routes[route.uri_template].add(route.method_name) for uri_template, methods_names in routes.items(): if not 'OPTIONS' in methods_names: options_operation = create_default_options(methods_names) uri_template_norm = uri_template.replace('{', '_').replace('}', '_') options_operation_name = '{}_{}'.format(uri_template_norm, 'options') setattr(cls, options_operation_name, options_operation) route = Route(uri_template, 'OPTIONS', options_operation_name, cls, {}, [], cls.__authorizer__) cls.__options_routes__.add(route) cls.__routes__.add(route)
