Module hug.routing
hug/routing.py
Defines the chainable classes responsible for defining the routing of Python functions for use with Falcon and CLIs
Copyright (C) 2016 Timothy Edmund Crosley
Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
View Source
"""hug/routing.py Defines the chainable classes responsible for defining the routing of Python functions for use with Falcon and CLIs Copyright (C) 2016 Timothy Edmund Crosley Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. """ from __future__ import absolute_import import os import re from collections import OrderedDict from functools import wraps from urllib.parse import urljoin import falcon from falcon import HTTP_METHODS import hug.api import hug.interface import hug.output_format from hug import introspect from hug.exceptions import InvalidTypeData class Router(object): """The base chainable router object""" __slots__ = ("route",) def __init__( self, transform=None, output=None, validate=None, api=None, requires=(), map_params=None, args=None, **kwargs ): self.route = {} if transform is not None: self.route["transform"] = transform if output: self.route["output"] = output if validate: self.route["validate"] = validate if api: self.route["api"] = api if requires: self.route["requires"] = ( (requires,) if not isinstance(requires, (tuple, list)) else requires ) if map_params: self.route["map_params"] = map_params if args: self.route["args"] = args def output(self, formatter, **overrides): """Sets the output formatter that should be used to render this route""" return self.where(output=formatter, **overrides) def transform(self, function, **overrides): """Sets the function that should be used to transform the returned Python structure into something serializable by specified output format """ return self.where(transform=function, **overrides) def validate(self, validation_function, **overrides): """Sets the secondary validation fucntion to use for this handler""" return self.where(validate=validation_function, **overrides) def api(self, api, **overrides): """Sets the API that should contain this route""" return self.where(api=api, **overrides) def requires(self, requirements, **overrides): """Adds additional requirements to the specified route""" return self.where( requires=tuple(self.route.get("requires", ())) + tuple(requirements), **overrides ) def doesnt_require(self, requirements, **overrides): """Removes individual requirements while keeping all other defined ones within a route""" return self.where( requires=tuple( set(self.route.get("requires", ())).difference( requirements if type(requirements) in (list, tuple) else (requirements,) ) ) ) def map_params(self, **map_params): """Map interface specific params to an internal name representation""" return self.where(map_params=map_params) def where(self, **overrides): """Creates a new route, based on the current route, with the specified overrided values""" route_data = self.route.copy() route_data.update(overrides) return self.__class__(**route_data) class CLIRouter(Router): """The CLIRouter provides a chainable router that can be used to route a CLI command to a Python function""" __slots__ = () def __init__(self, name=None, version=None, doc=None, **kwargs): super().__init__(**kwargs) if name is not None: self.route["name"] = name if version: self.route["version"] = version if doc: self.route["doc"] = doc def name(self, name, **overrides): """Sets the name for the CLI interface""" return self.where(name=name, **overrides) def version(self, version, **overrides): """Sets the version for the CLI interface""" return self.where(version=version, **overrides) def doc(self, documentation, **overrides): """Sets the documentation for the CLI interface""" return self.where(doc=documentation, **overrides) def __call__(self, api_function): """Enables exposing a Hug compatible function as a Command Line Interface""" hug.interface.CLI(self.route, api_function) return api_function class InternalValidation(Router): """Defines the base route for interfaces that define their own internal validation""" __slots__ = () def __init__(self, raise_on_invalid=False, on_invalid=None, output_invalid=None, **kwargs): super().__init__(**kwargs) if raise_on_invalid: self.route["raise_on_invalid"] = raise_on_invalid if on_invalid is not None: self.route["on_invalid"] = on_invalid if output_invalid is not None: self.route["output_invalid"] = output_invalid def raise_on_invalid(self, setting=True, **overrides): """Sets the route to raise validation errors instead of catching them""" return self.where(raise_on_invalid=setting, **overrides) def on_invalid(self, function, **overrides): """Sets a function to use to transform data on validation errors. Defaults to the transform function if one is set to ensure no special handling occurs for invalid data set to `False`. """ return self.where(on_invalid=function, **overrides) def output_invalid(self, output_handler, **overrides): """Sets an output handler to be used when handler validation fails. Defaults to the output formatter set globally for the route. """ return self.where(output_invalid=output_handler, **overrides) class LocalRouter(InternalValidation): """The LocalRouter defines how interfaces should be handled when accessed locally from within Python code""" __slots__ = () def __init__(self, directives=True, validate=True, version=None, **kwargs): super().__init__(**kwargs) if version is not None: self.route["version"] = version if not directives: self.route["skip_directives"] = True if not validate: self.route["skip_validation"] = True def directives(self, use=True, **kwargs): return self.where(directives=use) def validate(self, enforce=True, **kwargs): return self.where(validate=enforce) def version(self, supported, **kwargs): return self.where(version=supported) def __call__(self, api_function): """Enables exposing a hug compatible function locally""" return hug.interface.Local(self.route, api_function) class HTTPRouter(InternalValidation): """The HTTPRouter provides the base concept of a router from an HTTPRequest to a Python function""" __slots__ = () def __init__( self, versions=any, parse_body=False, parameters=None, defaults=None, status=None, response_headers=None, private=False, inputs=None, **kwargs ): if defaults is None: defaults = {} super().__init__(**kwargs) if versions is not any: self.route["versions"] = ( (versions,) if isinstance(versions, (int, float, None.__class__)) else versions ) self.route["versions"] = tuple( int(version) if version else version for version in self.route["versions"] ) if parse_body: self.route["parse_body"] = parse_body if parameters: self.route["parameters"] = parameters if defaults: self.route["defaults"] = defaults if status: self.route["status"] = status if response_headers: self.route["response_headers"] = response_headers if private: self.route["private"] = private if inputs: self.route["inputs"] = inputs def versions(self, supported, **overrides): """Sets the versions that this route should be compatiable with""" return self.where(versions=supported, **overrides) def parse_body(self, automatic=True, **overrides): """Tells hug to automatically parse the input body if it matches a registered input format""" return self.where(parse_body=automatic, **overrides) def set_status(self, status, **overrides): """Sets the status that will be returned by default""" return self.where(status=status, **overrides) def parameters(self, parameters, **overrides): """Sets the custom parameters that will be used instead of those found introspecting the decorated function""" return self.where(parameters=parameters, **overrides) def defaults(self, defaults, **overrides): """Sets the custom defaults that will be used for custom parameters""" return self.where(defaults=defaults, **overrides) def _create_interface(self, api, api_function, catch_exceptions=True): interface = hug.interface.HTTP(self.route, api_function, catch_exceptions) return (interface, api_function) def response_headers(self, headers, **overrides): """Sets the response headers automatically injected by the router""" return self.where(response_headers=headers, **overrides) def add_response_headers(self, headers, **overrides): """Adds the specified response headers while keeping existing ones in-tact""" response_headers = self.route.get("response_headers", {}).copy() response_headers.update(headers) return self.where(response_headers=response_headers, **overrides) def cache( self, private=False, max_age=31536000, s_maxage=None, no_cache=False, no_store=False, must_revalidate=False, **overrides ): """Convenience method for quickly adding cache header to route""" parts = ( "private" if private else "public", "max-age={0}".format(max_age), "s-maxage={0}".format(s_maxage) if s_maxage is not None else None, no_cache and "no-cache", no_store and "no-store", must_revalidate and "must-revalidate", ) return self.add_response_headers( {"cache-control": ", ".join(filter(bool, parts))}, **overrides ) def allow_origins( self, *origins, methods=None, max_age=None, credentials=None, headers=None, **overrides ): """Convenience method for quickly allowing other resources to access this one""" response_headers = {} if origins: @hug.response_middleware(api=self.route.get("api", None)) def process_data(request, response, resource): if "ORIGIN" in request.headers: origin = request.headers["ORIGIN"] if origin in origins: response.set_header("Access-Control-Allow-Origin", origin) else: response_headers["Access-Control-Allow-Origin"] = "*" if methods: response_headers["Access-Control-Allow-Methods"] = ", ".join(methods) if max_age: response_headers["Access-Control-Max-Age"] = max_age if credentials: response_headers["Access-Control-Allow-Credentials"] = str(credentials).lower() if headers: response_headers["Access-Control-Allow-Headers"] = headers return self.add_response_headers(response_headers, **overrides) class NotFoundRouter(HTTPRouter): """Provides a chainable router that can be used to route 404'd request to a Python function""" __slots__ = () def __init__(self, output=None, versions=any, status=falcon.HTTP_NOT_FOUND, **kwargs): super().__init__(output=output, versions=versions, status=status, **kwargs) def __call__(self, api_function): api = self.route.get("api", hug.api.from_object(api_function)) (interface, callable_method) = self._create_interface(api, api_function) for version in self.route.get("versions", (None,)): api.http.set_not_found_handler(interface, version) return callable_method class SinkRouter(HTTPRouter): """Provides a chainable router that can be used to route all routes pass a certain base URL (essentially route/*)""" __slots__ = () def __init__(self, urls=None, output=None, **kwargs): super().__init__(output=output, **kwargs) if urls: self.route["urls"] = (urls,) if isinstance(urls, str) else urls def __call__(self, api_function): api = self.route.get("api", hug.api.from_object(api_function)) (interface, callable_method) = self._create_interface(api, api_function) for base_url in self.route.get("urls", ("/{0}".format(api_function.__name__),)): api.http.add_sink(interface, base_url) return callable_method class StaticRouter(SinkRouter): """Provides a chainable router that can be used to return static files automatically from a set of directories""" __slots__ = ("route",) def __init__(self, urls=None, output=hug.output_format.file, cache=False, **kwargs): super().__init__(urls=urls, output=output, **kwargs) if cache is True: self.cache() elif cache is not False: self.cache(**cache) def __call__(self, api_function): directories = [] for directory in api_function(): path = os.path.abspath(directory) directories.append(path) api = self.route.get("api", hug.api.from_object(api_function)) for base_url in self.route.get("urls", ("/{0}".format(api_function.__name__),)): def read_file(request=None, path=""): filename = path.lstrip("/") for directory in directories: path = os.path.abspath(os.path.join(directory, filename)) if not path.startswith(directory): hug.redirect.not_found() if os.path.isdir(path): new_path = os.path.join(path, "index.html") if os.path.exists(new_path) and os.path.isfile(new_path): path = new_path if os.path.exists(path) and os.path.isfile(path): return path hug.redirect.not_found() api.http.add_sink(self._create_interface(api, read_file)[0], base_url) return api_function class ExceptionRouter(HTTPRouter): """Provides a chainable router that can be used to route exceptions thrown during request handling""" __slots__ = () def __init__(self, exceptions=(Exception,), exclude=(), output=None, **kwargs): super().__init__(output=output, **kwargs) self.route["exceptions"] = ( (exceptions,) if not isinstance(exceptions, (list, tuple)) else exceptions ) self.route["exclude"] = (exclude,) if not isinstance(exclude, (list, tuple)) else exclude def __call__(self, api_function): api = self.route.get("api", hug.api.from_object(api_function)) (interface, callable_method) = self._create_interface( api, api_function, catch_exceptions=False ) for version in self.route.get("versions", (None,)): for exception in self.route["exceptions"]: api.http.add_exception_handler(exception, interface, version) return callable_method def _create_interface(self, api, api_function, catch_exceptions=False): interface = hug.interface.ExceptionRaised(self.route, api_function, catch_exceptions) return (interface, api_function) class URLRouter(HTTPRouter): """Provides a chainable router that can be used to route a URL to a Python function""" __slots__ = () def __init__( self, urls=None, accept=HTTP_METHODS, output=None, examples=(), versions=any, suffixes=(), prefixes=(), response_headers=None, parse_body=True, **kwargs ): super().__init__( output=output, versions=versions, parse_body=parse_body, response_headers=response_headers, **kwargs ) if urls is not None: self.route["urls"] = (urls,) if isinstance(urls, str) else urls if accept: self.route["accept"] = (accept,) if isinstance(accept, str) else accept if examples: self.route["examples"] = (examples,) if isinstance(examples, str) else examples if suffixes: self.route["suffixes"] = (suffixes,) if isinstance(suffixes, str) else suffixes if prefixes: self.route["prefixes"] = (prefixes,) if isinstance(prefixes, str) else prefixes def __call__(self, api_function): api = self.route.get("api", hug.api.from_object(api_function)) api.http.routes.setdefault(api.http.base_url, OrderedDict()) (interface, callable_method) = self._create_interface(api, api_function) use_examples = self.route.get("examples", ()) if not interface.required and not use_examples: use_examples = (True,) for base_url in self.route.get("urls", ("/{0}".format(api_function.__name__),)): expose = [base_url] for suffix in self.route.get("suffixes", ()): if suffix.startswith("/"): expose.append(os.path.join(base_url, suffix.lstrip("/"))) else: expose.append(base_url + suffix) for prefix in self.route.get("prefixes", ()): expose.append(prefix + base_url) for url in expose: handlers = api.http.routes[api.http.base_url].setdefault(url, {}) for method in self.route.get("accept", ()): version_mapping = handlers.setdefault(method.upper(), {}) for version in self.route.get("versions", (None,)): version_mapping[version] = interface api.http.versioned.setdefault(version, {})[ callable_method.__name__ ] = callable_method interface.examples = use_examples return callable_method def urls(self, *urls, **overrides): """Sets the URLs that will map to this API call""" return self.where(urls=urls, **overrides) def accept(self, *accept, **overrides): """Sets a list of HTTP methods this router should accept""" return self.where(accept=accept, **overrides) def get(self, urls=None, **overrides): """Sets the acceptable HTTP method to a GET""" if urls is not None: overrides["urls"] = urls return self.where(accept="GET", **overrides) def delete(self, urls=None, **overrides): """Sets the acceptable HTTP method to DELETE""" if urls is not None: overrides["urls"] = urls return self.where(accept="DELETE", **overrides) def post(self, urls=None, **overrides): """Sets the acceptable HTTP method to POST""" if urls is not None: overrides["urls"] = urls return self.where(accept="POST", **overrides) def put(self, urls=None, **overrides): """Sets the acceptable HTTP method to PUT""" if urls is not None: overrides["urls"] = urls return self.where(accept="PUT", **overrides) def trace(self, urls=None, **overrides): """Sets the acceptable HTTP method to TRACE""" if urls is not None: overrides["urls"] = urls return self.where(accept="TRACE", **overrides) def patch(self, urls=None, **overrides): """Sets the acceptable HTTP method to PATCH""" if urls is not None: overrides["urls"] = urls return self.where(accept="PATCH", **overrides) def options(self, urls=None, **overrides): """Sets the acceptable HTTP method to OPTIONS""" if urls is not None: overrides["urls"] = urls return self.where(accept="OPTIONS", **overrides) def head(self, urls=None, **overrides): """Sets the acceptable HTTP method to HEAD""" if urls is not None: overrides["urls"] = urls return self.where(accept="HEAD", **overrides) def connect(self, urls=None, **overrides): """Sets the acceptable HTTP method to CONNECT""" if urls is not None: overrides["urls"] = urls return self.where(accept="CONNECT", **overrides) def call(self, **overrides): """Sets the acceptable HTTP method to all known""" return self.where(accept=HTTP_METHODS, **overrides) def http(self, **overrides): """Sets the acceptable HTTP method to all known""" return self.where(accept=HTTP_METHODS, **overrides) def get_post(self, **overrides): """Exposes a Python method externally under both the HTTP POST and GET methods""" return self.where(accept=("GET", "POST"), **overrides) def put_post(self, **overrides): """Exposes a Python method externally under both the HTTP POST and PUT methods""" return self.where(accept=("PUT", "POST"), **overrides) def examples(self, *examples, **overrides): """Sets the examples that the route should use""" return self.where(examples=examples, **overrides) def suffixes(self, *suffixes, **overrides): """Sets the suffixes supported by the route""" return self.where(suffixes=suffixes, **overrides) def prefixes(self, *prefixes, **overrides): """Sets the prefixes supported by the route""" return self.where(prefixes=prefixes, **overrides) def where(self, **overrides): if "urls" in overrides: existing_urls = self.route.get("urls", ()) use_urls = [] for url in ( (overrides["urls"],) if isinstance(overrides["urls"], str) else overrides["urls"] ): if url.startswith("/") or not existing_urls: use_urls.append(url) else: for existing in existing_urls: use_urls.append(urljoin(existing.rstrip("/") + "/", url)) overrides["urls"] = tuple(use_urls) return super().where(**overrides)
Variables
HTTP_METHODS
Classes
CLIRouter
class CLIRouter( name=None, version=None, doc=None, **kwargs )
The CLIRouter provides a chainable router that can be used to route a CLI command to a Python function
View Source
class CLIRouter(Router): """The CLIRouter provides a chainable router that can be used to route a CLI command to a Python function""" __slots__ = () def __init__(self, name=None, version=None, doc=None, **kwargs): super().__init__(**kwargs) if name is not None: self.route["name"] = name if version: self.route["version"] = version if doc: self.route["doc"] = doc def name(self, name, **overrides): """Sets the name for the CLI interface""" return self.where(name=name, **overrides) def version(self, version, **overrides): """Sets the version for the CLI interface""" return self.where(version=version, **overrides) def doc(self, documentation, **overrides): """Sets the documentation for the CLI interface""" return self.where(doc=documentation, **overrides) def __call__(self, api_function): """Enables exposing a Hug compatible function as a Command Line Interface""" hug.interface.CLI(self.route, api_function) return api_function
Ancestors (in MRO)
- hug.routing.Router
Class variables
route
Instance variables
route
Methods
api
def api( self, api, **overrides )
Sets the API that should contain this route
View Source
def api(self, api, **overrides): """Sets the API that should contain this route""" return self.where(api=api, **overrides)
doc
def doc( self, documentation, **overrides )
Sets the documentation for the CLI interface
View Source
def doc(self, documentation, **overrides): """Sets the documentation for the CLI interface""" return self.where(doc=documentation, **overrides)
doesnt_require
def doesnt_require( self, requirements, **overrides )
Removes individual requirements while keeping all other defined ones within a route
View Source
def doesnt_require(self, requirements, **overrides): """Removes individual requirements while keeping all other defined ones within a route""" return self.where( requires=tuple( set(self.route.get("requires", ())).difference( requirements if type(requirements) in (list, tuple) else (requirements,) ) ) )
map_params
def map_params( self, **map_params )
Map interface specific params to an internal name representation
View Source
def map_params(self, **map_params): """Map interface specific params to an internal name representation""" return self.where(map_params=map_params)
name
def name( self, name, **overrides )
Sets the name for the CLI interface
View Source
def name(self, name, **overrides): """Sets the name for the CLI interface""" return self.where(name=name, **overrides)
output
def output( self, formatter, **overrides )
Sets the output formatter that should be used to render this route
View Source
def output(self, formatter, **overrides): """Sets the output formatter that should be used to render this route""" return self.where(output=formatter, **overrides)
requires
def requires( self, requirements, **overrides )
Adds additional requirements to the specified route
View Source
def requires(self, requirements, **overrides): """Adds additional requirements to the specified route""" return self.where( requires=tuple(self.route.get("requires", ())) + tuple(requirements), **overrides )
transform
def transform( self, function, **overrides )
Sets the function that should be used to transform the returned Python structure into something serializable by specified output format
View Source
def transform(self, function, **overrides): """Sets the function that should be used to transform the returned Python structure into something serializable by specified output format """ return self.where(transform=function, **overrides)
validate
def validate( self, validation_function, **overrides )
Sets the secondary validation fucntion to use for this handler
View Source
def validate(self, validation_function, **overrides): """Sets the secondary validation fucntion to use for this handler""" return self.where(validate=validation_function, **overrides)
version
def version( self, version, **overrides )
Sets the version for the CLI interface
View Source
def version(self, version, **overrides): """Sets the version for the CLI interface""" return self.where(version=version, **overrides)
where
def where( self, **overrides )
Creates a new route, based on the current route, with the specified overrided values
View Source
def where(self, **overrides): """Creates a new route, based on the current route, with the specified overrided values""" route_data = self.route.copy() route_data.update(overrides) return self.__class__(**route_data)
ExceptionRouter
class ExceptionRouter( exceptions=(<class 'Exception'>,), exclude=(), output=None, **kwargs )
Provides a chainable router that can be used to route exceptions thrown during request handling
View Source
class ExceptionRouter(HTTPRouter): """Provides a chainable router that can be used to route exceptions thrown during request handling""" __slots__ = () def __init__(self, exceptions=(Exception,), exclude=(), output=None, **kwargs): super().__init__(output=output, **kwargs) self.route["exceptions"] = ( (exceptions,) if not isinstance(exceptions, (list, tuple)) else exceptions ) self.route["exclude"] = (exclude,) if not isinstance(exclude, (list, tuple)) else exclude def __call__(self, api_function): api = self.route.get("api", hug.api.from_object(api_function)) (interface, callable_method) = self._create_interface( api, api_function, catch_exceptions=False ) for version in self.route.get("versions", (None,)): for exception in self.route["exceptions"]: api.http.add_exception_handler(exception, interface, version) return callable_method def _create_interface(self, api, api_function, catch_exceptions=False): interface = hug.interface.ExceptionRaised(self.route, api_function, catch_exceptions) return (interface, api_function)
Ancestors (in MRO)
- hug.routing.HTTPRouter
- hug.routing.InternalValidation
- hug.routing.Router
Class variables
route
Instance variables
route
Methods
add_response_headers
def add_response_headers( self, headers, **overrides )
Adds the specified response headers while keeping existing ones in-tact
View Source
def add_response_headers(self, headers, **overrides): """Adds the specified response headers while keeping existing ones in-tact""" response_headers = self.route.get("response_headers", {}).copy() response_headers.update(headers) return self.where(response_headers=response_headers, **overrides)
allow_origins
def allow_origins( self, *origins, methods=None, max_age=None, credentials=None, headers=None, **overrides )
Convenience method for quickly allowing other resources to access this one
View Source
def allow_origins( self, *origins, methods=None, max_age=None, credentials=None, headers=None, **overrides ): """Convenience method for quickly allowing other resources to access this one""" response_headers = {} if origins: @hug.response_middleware(api=self.route.get("api", None)) def process_data(request, response, resource): if "ORIGIN" in request.headers: origin = request.headers["ORIGIN"] if origin in origins: response.set_header("Access-Control-Allow-Origin", origin) else: response_headers["Access-Control-Allow-Origin"] = "*" if methods: response_headers["Access-Control-Allow-Methods"] = ", ".join(methods) if max_age: response_headers["Access-Control-Max-Age"] = max_age if credentials: response_headers["Access-Control-Allow-Credentials"] = str(credentials).lower() if headers: response_headers["Access-Control-Allow-Headers"] = headers return self.add_response_headers(response_headers, **overrides)
api
def api( self, api, **overrides )
Sets the API that should contain this route
View Source
def api(self, api, **overrides): """Sets the API that should contain this route""" return self.where(api=api, **overrides)
cache
def cache( self, private=False, max_age=31536000, s_maxage=None, no_cache=False, no_store=False, must_revalidate=False, **overrides )
Convenience method for quickly adding cache header to route
View Source
def cache( self, private=False, max_age=31536000, s_maxage=None, no_cache=False, no_store=False, must_revalidate=False, **overrides ): """Convenience method for quickly adding cache header to route""" parts = ( "private" if private else "public", "max-age={0}".format(max_age), "s-maxage={0}".format(s_maxage) if s_maxage is not None else None, no_cache and "no-cache", no_store and "no-store", must_revalidate and "must-revalidate", ) return self.add_response_headers( {"cache-control": ", ".join(filter(bool, parts))}, **overrides )
defaults
def defaults( self, defaults, **overrides )
Sets the custom defaults that will be used for custom parameters
View Source
def defaults(self, defaults, **overrides): """Sets the custom defaults that will be used for custom parameters""" return self.where(defaults=defaults, **overrides)
doesnt_require
def doesnt_require( self, requirements, **overrides )
Removes individual requirements while keeping all other defined ones within a route
View Source
def doesnt_require(self, requirements, **overrides): """Removes individual requirements while keeping all other defined ones within a route""" return self.where( requires=tuple( set(self.route.get("requires", ())).difference( requirements if type(requirements) in (list, tuple) else (requirements,) ) ) )
map_params
def map_params( self, **map_params )
Map interface specific params to an internal name representation
View Source
def map_params(self, **map_params): """Map interface specific params to an internal name representation""" return self.where(map_params=map_params)
on_invalid
def on_invalid( self, function, **overrides )
Sets a function to use to transform data on validation errors.
Defaults to the transform function if one is set to ensure no special
handling occurs for invalid data set to False.
View Source
def on_invalid(self, function, **overrides): """Sets a function to use to transform data on validation errors. Defaults to the transform function if one is set to ensure no special handling occurs for invalid data set to `False`. """ return self.where(on_invalid=function, **overrides)
output
def output( self, formatter, **overrides )
Sets the output formatter that should be used to render this route
View Source
def output(self, formatter, **overrides): """Sets the output formatter that should be used to render this route""" return self.where(output=formatter, **overrides)
output_invalid
def output_invalid( self, output_handler, **overrides )
Sets an output handler to be used when handler validation fails.
Defaults to the output formatter set globally for the route.
View Source
def output_invalid(self, output_handler, **overrides): """Sets an output handler to be used when handler validation fails. Defaults to the output formatter set globally for the route. """ return self.where(output_invalid=output_handler, **overrides)
parameters
def parameters( self, parameters, **overrides )
Sets the custom parameters that will be used instead of those found introspecting the decorated function
View Source
def parameters(self, parameters, **overrides): """Sets the custom parameters that will be used instead of those found introspecting the decorated function""" return self.where(parameters=parameters, **overrides)
parse_body
def parse_body( self, automatic=True, **overrides )
Tells hug to automatically parse the input body if it matches a registered input format
View Source
def parse_body(self, automatic=True, **overrides): """Tells hug to automatically parse the input body if it matches a registered input format""" return self.where(parse_body=automatic, **overrides)
raise_on_invalid
def raise_on_invalid( self, setting=True, **overrides )
Sets the route to raise validation errors instead of catching them
View Source
def raise_on_invalid(self, setting=True, **overrides): """Sets the route to raise validation errors instead of catching them""" return self.where(raise_on_invalid=setting, **overrides)
requires
def requires( self, requirements, **overrides )
Adds additional requirements to the specified route
View Source
def requires(self, requirements, **overrides): """Adds additional requirements to the specified route""" return self.where( requires=tuple(self.route.get("requires", ())) + tuple(requirements), **overrides )
response_headers
def response_headers( self, headers, **overrides )
Sets the response headers automatically injected by the router
View Source
def response_headers(self, headers, **overrides): """Sets the response headers automatically injected by the router""" return self.where(response_headers=headers, **overrides)
set_status
def set_status( self, status, **overrides )
Sets the status that will be returned by default
View Source
def set_status(self, status, **overrides): """Sets the status that will be returned by default""" return self.where(status=status, **overrides)
transform
def transform( self, function, **overrides )
Sets the function that should be used to transform the returned Python structure into something serializable by specified output format
View Source
def transform(self, function, **overrides): """Sets the function that should be used to transform the returned Python structure into something serializable by specified output format """ return self.where(transform=function, **overrides)
validate
def validate( self, validation_function, **overrides )
Sets the secondary validation fucntion to use for this handler
View Source
def validate(self, validation_function, **overrides): """Sets the secondary validation fucntion to use for this handler""" return self.where(validate=validation_function, **overrides)
versions
def versions( self, supported, **overrides )
Sets the versions that this route should be compatiable with
View Source
def versions(self, supported, **overrides): """Sets the versions that this route should be compatiable with""" return self.where(versions=supported, **overrides)
where
def where( self, **overrides )
Creates a new route, based on the current route, with the specified overrided values
View Source
def where(self, **overrides): """Creates a new route, based on the current route, with the specified overrided values""" route_data = self.route.copy() route_data.update(overrides) return self.__class__(**route_data)
HTTPRouter
class HTTPRouter( versions=<built-in function any>, parse_body=False, parameters=None, defaults=None, status=None, response_headers=None, private=False, inputs=None, **kwargs )
The HTTPRouter provides the base concept of a router from an HTTPRequest to a Python function
View Source
class HTTPRouter(InternalValidation): """The HTTPRouter provides the base concept of a router from an HTTPRequest to a Python function""" __slots__ = () def __init__( self, versions=any, parse_body=False, parameters=None, defaults=None, status=None, response_headers=None, private=False, inputs=None, **kwargs ): if defaults is None: defaults = {} super().__init__(**kwargs) if versions is not any: self.route["versions"] = ( (versions,) if isinstance(versions, (int, float, None.__class__)) else versions ) self.route["versions"] = tuple( int(version) if version else version for version in self.route["versions"] ) if parse_body: self.route["parse_body"] = parse_body if parameters: self.route["parameters"] = parameters if defaults: self.route["defaults"] = defaults if status: self.route["status"] = status if response_headers: self.route["response_headers"] = response_headers if private: self.route["private"] = private if inputs: self.route["inputs"] = inputs def versions(self, supported, **overrides): """Sets the versions that this route should be compatiable with""" return self.where(versions=supported, **overrides) def parse_body(self, automatic=True, **overrides): """Tells hug to automatically parse the input body if it matches a registered input format""" return self.where(parse_body=automatic, **overrides) def set_status(self, status, **overrides): """Sets the status that will be returned by default""" return self.where(status=status, **overrides) def parameters(self, parameters, **overrides): """Sets the custom parameters that will be used instead of those found introspecting the decorated function""" return self.where(parameters=parameters, **overrides) def defaults(self, defaults, **overrides): """Sets the custom defaults that will be used for custom parameters""" return self.where(defaults=defaults, **overrides) def _create_interface(self, api, api_function, catch_exceptions=True): interface = hug.interface.HTTP(self.route, api_function, catch_exceptions) return (interface, api_function) def response_headers(self, headers, **overrides): """Sets the response headers automatically injected by the router""" return self.where(response_headers=headers, **overrides) def add_response_headers(self, headers, **overrides): """Adds the specified response headers while keeping existing ones in-tact""" response_headers = self.route.get("response_headers", {}).copy() response_headers.update(headers) return self.where(response_headers=response_headers, **overrides) def cache( self, private=False, max_age=31536000, s_maxage=None, no_cache=False, no_store=False, must_revalidate=False, **overrides ): """Convenience method for quickly adding cache header to route""" parts = ( "private" if private else "public", "max-age={0}".format(max_age), "s-maxage={0}".format(s_maxage) if s_maxage is not None else None, no_cache and "no-cache", no_store and "no-store", must_revalidate and "must-revalidate", ) return self.add_response_headers( {"cache-control": ", ".join(filter(bool, parts))}, **overrides ) def allow_origins( self, *origins, methods=None, max_age=None, credentials=None, headers=None, **overrides ): """Convenience method for quickly allowing other resources to access this one""" response_headers = {} if origins: @hug.response_middleware(api=self.route.get("api", None)) def process_data(request, response, resource): if "ORIGIN" in request.headers: origin = request.headers["ORIGIN"] if origin in origins: response.set_header("Access-Control-Allow-Origin", origin) else: response_headers["Access-Control-Allow-Origin"] = "*" if methods: response_headers["Access-Control-Allow-Methods"] = ", ".join(methods) if max_age: response_headers["Access-Control-Max-Age"] = max_age if credentials: response_headers["Access-Control-Allow-Credentials"] = str(credentials).lower() if headers: response_headers["Access-Control-Allow-Headers"] = headers return self.add_response_headers(response_headers, **overrides)
Ancestors (in MRO)
- hug.routing.InternalValidation
- hug.routing.Router
Descendants
- hug.routing.NotFoundRouter
- hug.routing.SinkRouter
- hug.routing.ExceptionRouter
- hug.routing.URLRouter
Class variables
route
Instance variables
route
Methods
add_response_headers
def add_response_headers( self, headers, **overrides )
Adds the specified response headers while keeping existing ones in-tact
View Source
def add_response_headers(self, headers, **overrides): """Adds the specified response headers while keeping existing ones in-tact""" response_headers = self.route.get("response_headers", {}).copy() response_headers.update(headers) return self.where(response_headers=response_headers, **overrides)
allow_origins
def allow_origins( self, *origins, methods=None, max_age=None, credentials=None, headers=None, **overrides )
Convenience method for quickly allowing other resources to access this one
View Source
def allow_origins( self, *origins, methods=None, max_age=None, credentials=None, headers=None, **overrides ): """Convenience method for quickly allowing other resources to access this one""" response_headers = {} if origins: @hug.response_middleware(api=self.route.get("api", None)) def process_data(request, response, resource): if "ORIGIN" in request.headers: origin = request.headers["ORIGIN"] if origin in origins: response.set_header("Access-Control-Allow-Origin", origin) else: response_headers["Access-Control-Allow-Origin"] = "*" if methods: response_headers["Access-Control-Allow-Methods"] = ", ".join(methods) if max_age: response_headers["Access-Control-Max-Age"] = max_age if credentials: response_headers["Access-Control-Allow-Credentials"] = str(credentials).lower() if headers: response_headers["Access-Control-Allow-Headers"] = headers return self.add_response_headers(response_headers, **overrides)
api
def api( self, api, **overrides )
Sets the API that should contain this route
View Source
def api(self, api, **overrides): """Sets the API that should contain this route""" return self.where(api=api, **overrides)
cache
def cache( self, private=False, max_age=31536000, s_maxage=None, no_cache=False, no_store=False, must_revalidate=False, **overrides )
Convenience method for quickly adding cache header to route
View Source
def cache( self, private=False, max_age=31536000, s_maxage=None, no_cache=False, no_store=False, must_revalidate=False, **overrides ): """Convenience method for quickly adding cache header to route""" parts = ( "private" if private else "public", "max-age={0}".format(max_age), "s-maxage={0}".format(s_maxage) if s_maxage is not None else None, no_cache and "no-cache", no_store and "no-store", must_revalidate and "must-revalidate", ) return self.add_response_headers( {"cache-control": ", ".join(filter(bool, parts))}, **overrides )
defaults
def defaults( self, defaults, **overrides )
Sets the custom defaults that will be used for custom parameters
View Source
def defaults(self, defaults, **overrides): """Sets the custom defaults that will be used for custom parameters""" return self.where(defaults=defaults, **overrides)
doesnt_require
def doesnt_require( self, requirements, **overrides )
Removes individual requirements while keeping all other defined ones within a route
View Source
def doesnt_require(self, requirements, **overrides): """Removes individual requirements while keeping all other defined ones within a route""" return self.where( requires=tuple( set(self.route.get("requires", ())).difference( requirements if type(requirements) in (list, tuple) else (requirements,) ) ) )
map_params
def map_params( self, **map_params )
Map interface specific params to an internal name representation
View Source
def map_params(self, **map_params): """Map interface specific params to an internal name representation""" return self.where(map_params=map_params)
on_invalid
def on_invalid( self, function, **overrides )
Sets a function to use to transform data on validation errors.
Defaults to the transform function if one is set to ensure no special
handling occurs for invalid data set to False.
View Source
def on_invalid(self, function, **overrides): """Sets a function to use to transform data on validation errors. Defaults to the transform function if one is set to ensure no special handling occurs for invalid data set to `False`. """ return self.where(on_invalid=function, **overrides)
output
def output( self, formatter, **overrides )
Sets the output formatter that should be used to render this route
View Source
def output(self, formatter, **overrides): """Sets the output formatter that should be used to render this route""" return self.where(output=formatter, **overrides)
output_invalid
def output_invalid( self, output_handler, **overrides )
Sets an output handler to be used when handler validation fails.
Defaults to the output formatter set globally for the route.
View Source
def output_invalid(self, output_handler, **overrides): """Sets an output handler to be used when handler validation fails. Defaults to the output formatter set globally for the route. """ return self.where(output_invalid=output_handler, **overrides)
parameters
def parameters( self, parameters, **overrides )
Sets the custom parameters that will be used instead of those found introspecting the decorated function
View Source
def parameters(self, parameters, **overrides): """Sets the custom parameters that will be used instead of those found introspecting the decorated function""" return self.where(parameters=parameters, **overrides)
parse_body
def parse_body( self, automatic=True, **overrides )
Tells hug to automatically parse the input body if it matches a registered input format
View Source
def parse_body(self, automatic=True, **overrides): """Tells hug to automatically parse the input body if it matches a registered input format""" return self.where(parse_body=automatic, **overrides)
raise_on_invalid
def raise_on_invalid( self, setting=True, **overrides )
Sets the route to raise validation errors instead of catching them
View Source
def raise_on_invalid(self, setting=True, **overrides): """Sets the route to raise validation errors instead of catching them""" return self.where(raise_on_invalid=setting, **overrides)
requires
def requires( self, requirements, **overrides )
Adds additional requirements to the specified route
View Source
def requires(self, requirements, **overrides): """Adds additional requirements to the specified route""" return self.where( requires=tuple(self.route.get("requires", ())) + tuple(requirements), **overrides )
response_headers
def response_headers( self, headers, **overrides )
Sets the response headers automatically injected by the router
View Source
def response_headers(self, headers, **overrides): """Sets the response headers automatically injected by the router""" return self.where(response_headers=headers, **overrides)
set_status
def set_status( self, status, **overrides )
Sets the status that will be returned by default
View Source
def set_status(self, status, **overrides): """Sets the status that will be returned by default""" return self.where(status=status, **overrides)
transform
def transform( self, function, **overrides )
Sets the function that should be used to transform the returned Python structure into something serializable by specified output format
View Source
def transform(self, function, **overrides): """Sets the function that should be used to transform the returned Python structure into something serializable by specified output format """ return self.where(transform=function, **overrides)
validate
def validate( self, validation_function, **overrides )
Sets the secondary validation fucntion to use for this handler
View Source
def validate(self, validation_function, **overrides): """Sets the secondary validation fucntion to use for this handler""" return self.where(validate=validation_function, **overrides)
versions
def versions( self, supported, **overrides )
Sets the versions that this route should be compatiable with
View Source
def versions(self, supported, **overrides): """Sets the versions that this route should be compatiable with""" return self.where(versions=supported, **overrides)
where
def where( self, **overrides )
Creates a new route, based on the current route, with the specified overrided values
View Source
def where(self, **overrides): """Creates a new route, based on the current route, with the specified overrided values""" route_data = self.route.copy() route_data.update(overrides) return self.__class__(**route_data)
InternalValidation
class InternalValidation( raise_on_invalid=False, on_invalid=None, output_invalid=None, **kwargs )
Defines the base route for interfaces that define their own internal validation
View Source
class InternalValidation(Router): """Defines the base route for interfaces that define their own internal validation""" __slots__ = () def __init__(self, raise_on_invalid=False, on_invalid=None, output_invalid=None, **kwargs): super().__init__(**kwargs) if raise_on_invalid: self.route["raise_on_invalid"] = raise_on_invalid if on_invalid is not None: self.route["on_invalid"] = on_invalid if output_invalid is not None: self.route["output_invalid"] = output_invalid def raise_on_invalid(self, setting=True, **overrides): """Sets the route to raise validation errors instead of catching them""" return self.where(raise_on_invalid=setting, **overrides) def on_invalid(self, function, **overrides): """Sets a function to use to transform data on validation errors. Defaults to the transform function if one is set to ensure no special handling occurs for invalid data set to `False`. """ return self.where(on_invalid=function, **overrides) def output_invalid(self, output_handler, **overrides): """Sets an output handler to be used when handler validation fails. Defaults to the output formatter set globally for the route. """ return self.where(output_invalid=output_handler, **overrides)
Ancestors (in MRO)
- hug.routing.Router
Descendants
- hug.routing.LocalRouter
- hug.routing.HTTPRouter
Class variables
route
Instance variables
route
Methods
api
def api( self, api, **overrides )
Sets the API that should contain this route
View Source
def api(self, api, **overrides): """Sets the API that should contain this route""" return self.where(api=api, **overrides)
doesnt_require
def doesnt_require( self, requirements, **overrides )
Removes individual requirements while keeping all other defined ones within a route
View Source
def doesnt_require(self, requirements, **overrides): """Removes individual requirements while keeping all other defined ones within a route""" return self.where( requires=tuple( set(self.route.get("requires", ())).difference( requirements if type(requirements) in (list, tuple) else (requirements,) ) ) )
map_params
def map_params( self, **map_params )
Map interface specific params to an internal name representation
View Source
def map_params(self, **map_params): """Map interface specific params to an internal name representation""" return self.where(map_params=map_params)
on_invalid
def on_invalid( self, function, **overrides )
Sets a function to use to transform data on validation errors.
Defaults to the transform function if one is set to ensure no special
handling occurs for invalid data set to False.
View Source
def on_invalid(self, function, **overrides): """Sets a function to use to transform data on validation errors. Defaults to the transform function if one is set to ensure no special handling occurs for invalid data set to `False`. """ return self.where(on_invalid=function, **overrides)
output
def output( self, formatter, **overrides )
Sets the output formatter that should be used to render this route
View Source
def output(self, formatter, **overrides): """Sets the output formatter that should be used to render this route""" return self.where(output=formatter, **overrides)
output_invalid
def output_invalid( self, output_handler, **overrides )
Sets an output handler to be used when handler validation fails.
Defaults to the output formatter set globally for the route.
View Source
def output_invalid(self, output_handler, **overrides): """Sets an output handler to be used when handler validation fails. Defaults to the output formatter set globally for the route. """ return self.where(output_invalid=output_handler, **overrides)
raise_on_invalid
def raise_on_invalid( self, setting=True, **overrides )
Sets the route to raise validation errors instead of catching them
View Source
def raise_on_invalid(self, setting=True, **overrides): """Sets the route to raise validation errors instead of catching them""" return self.where(raise_on_invalid=setting, **overrides)
requires
def requires( self, requirements, **overrides )
Adds additional requirements to the specified route
View Source
def requires(self, requirements, **overrides): """Adds additional requirements to the specified route""" return self.where( requires=tuple(self.route.get("requires", ())) + tuple(requirements), **overrides )
transform
def transform( self, function, **overrides )
Sets the function that should be used to transform the returned Python structure into something serializable by specified output format
View Source
def transform(self, function, **overrides): """Sets the function that should be used to transform the returned Python structure into something serializable by specified output format """ return self.where(transform=function, **overrides)
validate
def validate( self, validation_function, **overrides )
Sets the secondary validation fucntion to use for this handler
View Source
def validate(self, validation_function, **overrides): """Sets the secondary validation fucntion to use for this handler""" return self.where(validate=validation_function, **overrides)
where
def where( self, **overrides )
Creates a new route, based on the current route, with the specified overrided values
View Source
def where(self, **overrides): """Creates a new route, based on the current route, with the specified overrided values""" route_data = self.route.copy() route_data.update(overrides) return self.__class__(**route_data)
LocalRouter
class LocalRouter( directives=True, validate=True, version=None, **kwargs )
The LocalRouter defines how interfaces should be handled when accessed locally from within Python code
View Source
class LocalRouter(InternalValidation): """The LocalRouter defines how interfaces should be handled when accessed locally from within Python code""" __slots__ = () def __init__(self, directives=True, validate=True, version=None, **kwargs): super().__init__(**kwargs) if version is not None: self.route["version"] = version if not directives: self.route["skip_directives"] = True if not validate: self.route["skip_validation"] = True def directives(self, use=True, **kwargs): return self.where(directives=use) def validate(self, enforce=True, **kwargs): return self.where(validate=enforce) def version(self, supported, **kwargs): return self.where(version=supported) def __call__(self, api_function): """Enables exposing a hug compatible function locally""" return hug.interface.Local(self.route, api_function)
Ancestors (in MRO)
- hug.routing.InternalValidation
- hug.routing.Router
Class variables
route
Instance variables
route
Methods
api
def api( self, api, **overrides )
Sets the API that should contain this route
View Source
def api(self, api, **overrides): """Sets the API that should contain this route""" return self.where(api=api, **overrides)
directives
def directives( self, use=True, **kwargs )
View Source
def directives(self, use=True, **kwargs): return self.where(directives=use)
doesnt_require
def doesnt_require( self, requirements, **overrides )
Removes individual requirements while keeping all other defined ones within a route
View Source
def doesnt_require(self, requirements, **overrides): """Removes individual requirements while keeping all other defined ones within a route""" return self.where( requires=tuple( set(self.route.get("requires", ())).difference( requirements if type(requirements) in (list, tuple) else (requirements,) ) ) )
map_params
def map_params( self, **map_params )
Map interface specific params to an internal name representation
View Source
def map_params(self, **map_params): """Map interface specific params to an internal name representation""" return self.where(map_params=map_params)
on_invalid
def on_invalid( self, function, **overrides )
Sets a function to use to transform data on validation errors.
Defaults to the transform function if one is set to ensure no special
handling occurs for invalid data set to False.
View Source
def on_invalid(self, function, **overrides): """Sets a function to use to transform data on validation errors. Defaults to the transform function if one is set to ensure no special handling occurs for invalid data set to `False`. """ return self.where(on_invalid=function, **overrides)
output
def output( self, formatter, **overrides )
Sets the output formatter that should be used to render this route
View Source
def output(self, formatter, **overrides): """Sets the output formatter that should be used to render this route""" return self.where(output=formatter, **overrides)
output_invalid
def output_invalid( self, output_handler, **overrides )
Sets an output handler to be used when handler validation fails.
Defaults to the output formatter set globally for the route.
View Source
def output_invalid(self, output_handler, **overrides): """Sets an output handler to be used when handler validation fails. Defaults to the output formatter set globally for the route. """ return self.where(output_invalid=output_handler, **overrides)
raise_on_invalid
def raise_on_invalid( self, setting=True, **overrides )
Sets the route to raise validation errors instead of catching them
View Source
def raise_on_invalid(self, setting=True, **overrides): """Sets the route to raise validation errors instead of catching them""" return self.where(raise_on_invalid=setting, **overrides)
requires
def requires( self, requirements, **overrides )
Adds additional requirements to the specified route
View Source
def requires(self, requirements, **overrides): """Adds additional requirements to the specified route""" return self.where( requires=tuple(self.route.get("requires", ())) + tuple(requirements), **overrides )
transform
def transform( self, function, **overrides )
Sets the function that should be used to transform the returned Python structure into something serializable by specified output format
View Source
def transform(self, function, **overrides): """Sets the function that should be used to transform the returned Python structure into something serializable by specified output format """ return self.where(transform=function, **overrides)
validate
def validate( self, enforce=True, **kwargs )
Sets the secondary validation fucntion to use for this handler
View Source
def validate(self, enforce=True, **kwargs): return self.where(validate=enforce)
version
def version( self, supported, **kwargs )
View Source
def version(self, supported, **kwargs): return self.where(version=supported)
where
def where( self, **overrides )
Creates a new route, based on the current route, with the specified overrided values
View Source
def where(self, **overrides): """Creates a new route, based on the current route, with the specified overrided values""" route_data = self.route.copy() route_data.update(overrides) return self.__class__(**route_data)
NotFoundRouter
class NotFoundRouter( output=None, versions=<built-in function any>, status='404 Not Found', **kwargs )
Provides a chainable router that can be used to route 404'd request to a Python function
View Source
class NotFoundRouter(HTTPRouter): """Provides a chainable router that can be used to route 404'd request to a Python function""" __slots__ = () def __init__(self, output=None, versions=any, status=falcon.HTTP_NOT_FOUND, **kwargs): super().__init__(output=output, versions=versions, status=status, **kwargs) def __call__(self, api_function): api = self.route.get("api", hug.api.from_object(api_function)) (interface, callable_method) = self._create_interface(api, api_function) for version in self.route.get("versions", (None,)): api.http.set_not_found_handler(interface, version) return callable_method
Ancestors (in MRO)
- hug.routing.HTTPRouter
- hug.routing.InternalValidation
- hug.routing.Router
Class variables
route
Instance variables
route
Methods
add_response_headers
def add_response_headers( self, headers, **overrides )
Adds the specified response headers while keeping existing ones in-tact
View Source
def add_response_headers(self, headers, **overrides): """Adds the specified response headers while keeping existing ones in-tact""" response_headers = self.route.get("response_headers", {}).copy() response_headers.update(headers) return self.where(response_headers=response_headers, **overrides)
allow_origins
def allow_origins( self, *origins, methods=None, max_age=None, credentials=None, headers=None, **overrides )
Convenience method for quickly allowing other resources to access this one
View Source
def allow_origins( self, *origins, methods=None, max_age=None, credentials=None, headers=None, **overrides ): """Convenience method for quickly allowing other resources to access this one""" response_headers = {} if origins: @hug.response_middleware(api=self.route.get("api", None)) def process_data(request, response, resource): if "ORIGIN" in request.headers: origin = request.headers["ORIGIN"] if origin in origins: response.set_header("Access-Control-Allow-Origin", origin) else: response_headers["Access-Control-Allow-Origin"] = "*" if methods: response_headers["Access-Control-Allow-Methods"] = ", ".join(methods) if max_age: response_headers["Access-Control-Max-Age"] = max_age if credentials: response_headers["Access-Control-Allow-Credentials"] = str(credentials).lower() if headers: response_headers["Access-Control-Allow-Headers"] = headers return self.add_response_headers(response_headers, **overrides)
api
def api( self, api, **overrides )
Sets the API that should contain this route
View Source
def api(self, api, **overrides): """Sets the API that should contain this route""" return self.where(api=api, **overrides)
cache
def cache( self, private=False, max_age=31536000, s_maxage=None, no_cache=False, no_store=False, must_revalidate=False, **overrides )
Convenience method for quickly adding cache header to route
View Source
def cache( self, private=False, max_age=31536000, s_maxage=None, no_cache=False, no_store=False, must_revalidate=False, **overrides ): """Convenience method for quickly adding cache header to route""" parts = ( "private" if private else "public", "max-age={0}".format(max_age), "s-maxage={0}".format(s_maxage) if s_maxage is not None else None, no_cache and "no-cache", no_store and "no-store", must_revalidate and "must-revalidate", ) return self.add_response_headers( {"cache-control": ", ".join(filter(bool, parts))}, **overrides )
defaults
def defaults( self, defaults, **overrides )
Sets the custom defaults that will be used for custom parameters
View Source
def defaults(self, defaults, **overrides): """Sets the custom defaults that will be used for custom parameters""" return self.where(defaults=defaults, **overrides)
doesnt_require
def doesnt_require( self, requirements, **overrides )
Removes individual requirements while keeping all other defined ones within a route
View Source
def doesnt_require(self, requirements, **overrides): """Removes individual requirements while keeping all other defined ones within a route""" return self.where( requires=tuple( set(self.route.get("requires", ())).difference( requirements if type(requirements) in (list, tuple) else (requirements,) ) ) )
map_params
def map_params( self, **map_params )
Map interface specific params to an internal name representation
View Source
def map_params(self, **map_params): """Map interface specific params to an internal name representation""" return self.where(map_params=map_params)
on_invalid
def on_invalid( self, function, **overrides )
Sets a function to use to transform data on validation errors.
Defaults to the transform function if one is set to ensure no special
handling occurs for invalid data set to False.
View Source
def on_invalid(self, function, **overrides): """Sets a function to use to transform data on validation errors. Defaults to the transform function if one is set to ensure no special handling occurs for invalid data set to `False`. """ return self.where(on_invalid=function, **overrides)
output
def output( self, formatter, **overrides )
Sets the output formatter that should be used to render this route
View Source
def output(self, formatter, **overrides): """Sets the output formatter that should be used to render this route""" return self.where(output=formatter, **overrides)
output_invalid
def output_invalid( self, output_handler, **overrides )
Sets an output handler to be used when handler validation fails.
Defaults to the output formatter set globally for the route.
View Source
def output_invalid(self, output_handler, **overrides): """Sets an output handler to be used when handler validation fails. Defaults to the output formatter set globally for the route. """ return self.where(output_invalid=output_handler, **overrides)
parameters
def parameters( self, parameters, **overrides )
Sets the custom parameters that will be used instead of those found introspecting the decorated function
View Source
def parameters(self, parameters, **overrides): """Sets the custom parameters that will be used instead of those found introspecting the decorated function""" return self.where(parameters=parameters, **overrides)
parse_body
def parse_body( self, automatic=True, **overrides )
Tells hug to automatically parse the input body if it matches a registered input format
View Source
def parse_body(self, automatic=True, **overrides): """Tells hug to automatically parse the input body if it matches a registered input format""" return self.where(parse_body=automatic, **overrides)
raise_on_invalid
def raise_on_invalid( self, setting=True, **overrides )
Sets the route to raise validation errors instead of catching them
View Source
def raise_on_invalid(self, setting=True, **overrides): """Sets the route to raise validation errors instead of catching them""" return self.where(raise_on_invalid=setting, **overrides)
requires
def requires( self, requirements, **overrides )
Adds additional requirements to the specified route
View Source
def requires(self, requirements, **overrides): """Adds additional requirements to the specified route""" return self.where( requires=tuple(self.route.get("requires", ())) + tuple(requirements), **overrides )
response_headers
def response_headers( self, headers, **overrides )
Sets the response headers automatically injected by the router
View Source
def response_headers(self, headers, **overrides): """Sets the response headers automatically injected by the router""" return self.where(response_headers=headers, **overrides)
set_status
def set_status( self, status, **overrides )
Sets the status that will be returned by default
View Source
def set_status(self, status, **overrides): """Sets the status that will be returned by default""" return self.where(status=status, **overrides)
transform
def transform( self, function, **overrides )
Sets the function that should be used to transform the returned Python structure into something serializable by specified output format
View Source
def transform(self, function, **overrides): """Sets the function that should be used to transform the returned Python structure into something serializable by specified output format """ return self.where(transform=function, **overrides)
validate
def validate( self, validation_function, **overrides )
Sets the secondary validation fucntion to use for this handler
View Source
def validate(self, validation_function, **overrides): """Sets the secondary validation fucntion to use for this handler""" return self.where(validate=validation_function, **overrides)
versions
def versions( self, supported, **overrides )
Sets the versions that this route should be compatiable with
View Source
def versions(self, supported, **overrides): """Sets the versions that this route should be compatiable with""" return self.where(versions=supported, **overrides)
where
def where( self, **overrides )
Creates a new route, based on the current route, with the specified overrided values
View Source
def where(self, **overrides): """Creates a new route, based on the current route, with the specified overrided values""" route_data = self.route.copy() route_data.update(overrides) return self.__class__(**route_data)
Router
class Router( transform=None, output=None, validate=None, api=None, requires=(), map_params=None, args=None, **kwargs )
The base chainable router object
View Source
class Router(object): """The base chainable router object""" __slots__ = ("route",) def __init__( self, transform=None, output=None, validate=None, api=None, requires=(), map_params=None, args=None, **kwargs ): self.route = {} if transform is not None: self.route["transform"] = transform if output: self.route["output"] = output if validate: self.route["validate"] = validate if api: self.route["api"] = api if requires: self.route["requires"] = ( (requires,) if not isinstance(requires, (tuple, list)) else requires ) if map_params: self.route["map_params"] = map_params if args: self.route["args"] = args def output(self, formatter, **overrides): """Sets the output formatter that should be used to render this route""" return self.where(output=formatter, **overrides) def transform(self, function, **overrides): """Sets the function that should be used to transform the returned Python structure into something serializable by specified output format """ return self.where(transform=function, **overrides) def validate(self, validation_function, **overrides): """Sets the secondary validation fucntion to use for this handler""" return self.where(validate=validation_function, **overrides) def api(self, api, **overrides): """Sets the API that should contain this route""" return self.where(api=api, **overrides) def requires(self, requirements, **overrides): """Adds additional requirements to the specified route""" return self.where( requires=tuple(self.route.get("requires", ())) + tuple(requirements), **overrides ) def doesnt_require(self, requirements, **overrides): """Removes individual requirements while keeping all other defined ones within a route""" return self.where( requires=tuple( set(self.route.get("requires", ())).difference( requirements if type(requirements) in (list, tuple) else (requirements,) ) ) ) def map_params(self, **map_params): """Map interface specific params to an internal name representation""" return self.where(map_params=map_params) def where(self, **overrides): """Creates a new route, based on the current route, with the specified overrided values""" route_data = self.route.copy() route_data.update(overrides) return self.__class__(**route_data)
Descendants
- hug.routing.CLIRouter
- hug.routing.InternalValidation
Instance variables
route
Methods
api
def api( self, api, **overrides )
Sets the API that should contain this route
View Source
def api(self, api, **overrides): """Sets the API that should contain this route""" return self.where(api=api, **overrides)
doesnt_require
def doesnt_require( self, requirements, **overrides )
Removes individual requirements while keeping all other defined ones within a route
View Source
def doesnt_require(self, requirements, **overrides): """Removes individual requirements while keeping all other defined ones within a route""" return self.where( requires=tuple( set(self.route.get("requires", ())).difference( requirements if type(requirements) in (list, tuple) else (requirements,) ) ) )
map_params
def map_params( self, **map_params )
Map interface specific params to an internal name representation
View Source
def map_params(self, **map_params): """Map interface specific params to an internal name representation""" return self.where(map_params=map_params)
output
def output( self, formatter, **overrides )
Sets the output formatter that should be used to render this route
View Source
def output(self, formatter, **overrides): """Sets the output formatter that should be used to render this route""" return self.where(output=formatter, **overrides)
requires
def requires( self, requirements, **overrides )
Adds additional requirements to the specified route
View Source
def requires(self, requirements, **overrides): """Adds additional requirements to the specified route""" return self.where( requires=tuple(self.route.get("requires", ())) + tuple(requirements), **overrides )
transform
def transform( self, function, **overrides )
Sets the function that should be used to transform the returned Python structure into something serializable by specified output format
View Source
def transform(self, function, **overrides): """Sets the function that should be used to transform the returned Python structure into something serializable by specified output format """ return self.where(transform=function, **overrides)
validate
def validate( self, validation_function, **overrides )
Sets the secondary validation fucntion to use for this handler
View Source
def validate(self, validation_function, **overrides): """Sets the secondary validation fucntion to use for this handler""" return self.where(validate=validation_function, **overrides)
where
def where( self, **overrides )
Creates a new route, based on the current route, with the specified overrided values
View Source
def where(self, **overrides): """Creates a new route, based on the current route, with the specified overrided values""" route_data = self.route.copy() route_data.update(overrides) return self.__class__(**route_data)
SinkRouter
class SinkRouter( urls=None, output=None, **kwargs )
Provides a chainable router that can be used to route all routes pass a certain base URL (essentially route/*)
View Source
class SinkRouter(HTTPRouter): """Provides a chainable router that can be used to route all routes pass a certain base URL (essentially route/*)""" __slots__ = () def __init__(self, urls=None, output=None, **kwargs): super().__init__(output=output, **kwargs) if urls: self.route["urls"] = (urls,) if isinstance(urls, str) else urls def __call__(self, api_function): api = self.route.get("api", hug.api.from_object(api_function)) (interface, callable_method) = self._create_interface(api, api_function) for base_url in self.route.get("urls", ("/{0}".format(api_function.__name__),)): api.http.add_sink(interface, base_url) return callable_method
Ancestors (in MRO)
- hug.routing.HTTPRouter
- hug.routing.InternalValidation
- hug.routing.Router
Descendants
- hug.routing.StaticRouter
Class variables
route
Instance variables
route
Methods
add_response_headers
def add_response_headers( self, headers, **overrides )
Adds the specified response headers while keeping existing ones in-tact
View Source
def add_response_headers(self, headers, **overrides): """Adds the specified response headers while keeping existing ones in-tact""" response_headers = self.route.get("response_headers", {}).copy() response_headers.update(headers) return self.where(response_headers=response_headers, **overrides)
allow_origins
def allow_origins( self, *origins, methods=None, max_age=None, credentials=None, headers=None, **overrides )
Convenience method for quickly allowing other resources to access this one
View Source
def allow_origins( self, *origins, methods=None, max_age=None, credentials=None, headers=None, **overrides ): """Convenience method for quickly allowing other resources to access this one""" response_headers = {} if origins: @hug.response_middleware(api=self.route.get("api", None)) def process_data(request, response, resource): if "ORIGIN" in request.headers: origin = request.headers["ORIGIN"] if origin in origins: response.set_header("Access-Control-Allow-Origin", origin) else: response_headers["Access-Control-Allow-Origin"] = "*" if methods: response_headers["Access-Control-Allow-Methods"] = ", ".join(methods) if max_age: response_headers["Access-Control-Max-Age"] = max_age if credentials: response_headers["Access-Control-Allow-Credentials"] = str(credentials).lower() if headers: response_headers["Access-Control-Allow-Headers"] = headers return self.add_response_headers(response_headers, **overrides)
api
def api( self, api, **overrides )
Sets the API that should contain this route
View Source
def api(self, api, **overrides): """Sets the API that should contain this route""" return self.where(api=api, **overrides)
cache
def cache( self, private=False, max_age=31536000, s_maxage=None, no_cache=False, no_store=False, must_revalidate=False, **overrides )
Convenience method for quickly adding cache header to route
View Source
def cache( self, private=False, max_age=31536000, s_maxage=None, no_cache=False, no_store=False, must_revalidate=False, **overrides ): """Convenience method for quickly adding cache header to route""" parts = ( "private" if private else "public", "max-age={0}".format(max_age), "s-maxage={0}".format(s_maxage) if s_maxage is not None else None, no_cache and "no-cache", no_store and "no-store", must_revalidate and "must-revalidate", ) return self.add_response_headers( {"cache-control": ", ".join(filter(bool, parts))}, **overrides )
defaults
def defaults( self, defaults, **overrides )
Sets the custom defaults that will be used for custom parameters
View Source
def defaults(self, defaults, **overrides): """Sets the custom defaults that will be used for custom parameters""" return self.where(defaults=defaults, **overrides)
doesnt_require
def doesnt_require( self, requirements, **overrides )
Removes individual requirements while keeping all other defined ones within a route
View Source
def doesnt_require(self, requirements, **overrides): """Removes individual requirements while keeping all other defined ones within a route""" return self.where( requires=tuple( set(self.route.get("requires", ())).difference( requirements if type(requirements) in (list, tuple) else (requirements,) ) ) )
map_params
def map_params( self, **map_params )
Map interface specific params to an internal name representation
View Source
def map_params(self, **map_params): """Map interface specific params to an internal name representation""" return self.where(map_params=map_params)
on_invalid
def on_invalid( self, function, **overrides )
Sets a function to use to transform data on validation errors.
Defaults to the transform function if one is set to ensure no special
handling occurs for invalid data set to False.
View Source
def on_invalid(self, function, **overrides): """Sets a function to use to transform data on validation errors. Defaults to the transform function if one is set to ensure no special handling occurs for invalid data set to `False`. """ return self.where(on_invalid=function, **overrides)
output
def output( self, formatter, **overrides )
Sets the output formatter that should be used to render this route
View Source
def output(self, formatter, **overrides): """Sets the output formatter that should be used to render this route""" return self.where(output=formatter, **overrides)
output_invalid
def output_invalid( self, output_handler, **overrides )
Sets an output handler to be used when handler validation fails.
Defaults to the output formatter set globally for the route.
View Source
def output_invalid(self, output_handler, **overrides): """Sets an output handler to be used when handler validation fails. Defaults to the output formatter set globally for the route. """ return self.where(output_invalid=output_handler, **overrides)
parameters
def parameters( self, parameters, **overrides )
Sets the custom parameters that will be used instead of those found introspecting the decorated function
View Source
def parameters(self, parameters, **overrides): """Sets the custom parameters that will be used instead of those found introspecting the decorated function""" return self.where(parameters=parameters, **overrides)
parse_body
def parse_body( self, automatic=True, **overrides )
Tells hug to automatically parse the input body if it matches a registered input format
View Source
def parse_body(self, automatic=True, **overrides): """Tells hug to automatically parse the input body if it matches a registered input format""" return self.where(parse_body=automatic, **overrides)
raise_on_invalid
def raise_on_invalid( self, setting=True, **overrides )
Sets the route to raise validation errors instead of catching them
View Source
def raise_on_invalid(self, setting=True, **overrides): """Sets the route to raise validation errors instead of catching them""" return self.where(raise_on_invalid=setting, **overrides)
requires
def requires( self, requirements, **overrides )
Adds additional requirements to the specified route
View Source
def requires(self, requirements, **overrides): """Adds additional requirements to the specified route""" return self.where( requires=tuple(self.route.get("requires", ())) + tuple(requirements), **overrides )
response_headers
def response_headers( self, headers, **overrides )
Sets the response headers automatically injected by the router
View Source
def response_headers(self, headers, **overrides): """Sets the response headers automatically injected by the router""" return self.where(response_headers=headers, **overrides)
set_status
def set_status( self, status, **overrides )
Sets the status that will be returned by default
View Source
def set_status(self, status, **overrides): """Sets the status that will be returned by default""" return self.where(status=status, **overrides)
transform
def transform( self, function, **overrides )
Sets the function that should be used to transform the returned Python structure into something serializable by specified output format
View Source
def transform(self, function, **overrides): """Sets the function that should be used to transform the returned Python structure into something serializable by specified output format """ return self.where(transform=function, **overrides)
validate
def validate( self, validation_function, **overrides )
Sets the secondary validation fucntion to use for this handler
View Source
def validate(self, validation_function, **overrides): """Sets the secondary validation fucntion to use for this handler""" return self.where(validate=validation_function, **overrides)
versions
def versions( self, supported, **overrides )
Sets the versions that this route should be compatiable with
View Source
def versions(self, supported, **overrides): """Sets the versions that this route should be compatiable with""" return self.where(versions=supported, **overrides)
where
def where( self, **overrides )
Creates a new route, based on the current route, with the specified overrided values
View Source
def where(self, **overrides): """Creates a new route, based on the current route, with the specified overrided values""" route_data = self.route.copy() route_data.update(overrides) return self.__class__(**route_data)
StaticRouter
class StaticRouter( urls=None, output=<function file at 0x7f7ee31b1e18>, cache=False, **kwargs )
Provides a chainable router that can be used to return static files automatically from a set of directories
View Source
class StaticRouter(SinkRouter): """Provides a chainable router that can be used to return static files automatically from a set of directories""" __slots__ = ("route",) def __init__(self, urls=None, output=hug.output_format.file, cache=False, **kwargs): super().__init__(urls=urls, output=output, **kwargs) if cache is True: self.cache() elif cache is not False: self.cache(**cache) def __call__(self, api_function): directories = [] for directory in api_function(): path = os.path.abspath(directory) directories.append(path) api = self.route.get("api", hug.api.from_object(api_function)) for base_url in self.route.get("urls", ("/{0}".format(api_function.__name__),)): def read_file(request=None, path=""): filename = path.lstrip("/") for directory in directories: path = os.path.abspath(os.path.join(directory, filename)) if not path.startswith(directory): hug.redirect.not_found() if os.path.isdir(path): new_path = os.path.join(path, "index.html") if os.path.exists(new_path) and os.path.isfile(new_path): path = new_path if os.path.exists(path) and os.path.isfile(path): return path hug.redirect.not_found() api.http.add_sink(self._create_interface(api, read_file)[0], base_url) return api_function
Ancestors (in MRO)
- hug.routing.SinkRouter
- hug.routing.HTTPRouter
- hug.routing.InternalValidation
- hug.routing.Router
Instance variables
route
Methods
add_response_headers
def add_response_headers( self, headers, **overrides )
Adds the specified response headers while keeping existing ones in-tact
View Source
def add_response_headers(self, headers, **overrides): """Adds the specified response headers while keeping existing ones in-tact""" response_headers = self.route.get("response_headers", {}).copy() response_headers.update(headers) return self.where(response_headers=response_headers, **overrides)
allow_origins
def allow_origins( self, *origins, methods=None, max_age=None, credentials=None, headers=None, **overrides )
Convenience method for quickly allowing other resources to access this one
View Source
def allow_origins( self, *origins, methods=None, max_age=None, credentials=None, headers=None, **overrides ): """Convenience method for quickly allowing other resources to access this one""" response_headers = {} if origins: @hug.response_middleware(api=self.route.get("api", None)) def process_data(request, response, resource): if "ORIGIN" in request.headers: origin = request.headers["ORIGIN"] if origin in origins: response.set_header("Access-Control-Allow-Origin", origin) else: response_headers["Access-Control-Allow-Origin"] = "*" if methods: response_headers["Access-Control-Allow-Methods"] = ", ".join(methods) if max_age: response_headers["Access-Control-Max-Age"] = max_age if credentials: response_headers["Access-Control-Allow-Credentials"] = str(credentials).lower() if headers: response_headers["Access-Control-Allow-Headers"] = headers return self.add_response_headers(response_headers, **overrides)
api
def api( self, api, **overrides )
Sets the API that should contain this route
View Source
def api(self, api, **overrides): """Sets the API that should contain this route""" return self.where(api=api, **overrides)
cache
def cache( self, private=False, max_age=31536000, s_maxage=None, no_cache=False, no_store=False, must_revalidate=False, **overrides )
Convenience method for quickly adding cache header to route
View Source
def cache( self, private=False, max_age=31536000, s_maxage=None, no_cache=False, no_store=False, must_revalidate=False, **overrides ): """Convenience method for quickly adding cache header to route""" parts = ( "private" if private else "public", "max-age={0}".format(max_age), "s-maxage={0}".format(s_maxage) if s_maxage is not None else None, no_cache and "no-cache", no_store and "no-store", must_revalidate and "must-revalidate", ) return self.add_response_headers( {"cache-control": ", ".join(filter(bool, parts))}, **overrides )
defaults
def defaults( self, defaults, **overrides )
Sets the custom defaults that will be used for custom parameters
View Source
def defaults(self, defaults, **overrides): """Sets the custom defaults that will be used for custom parameters""" return self.where(defaults=defaults, **overrides)
doesnt_require
def doesnt_require( self, requirements, **overrides )
Removes individual requirements while keeping all other defined ones within a route
View Source
def doesnt_require(self, requirements, **overrides): """Removes individual requirements while keeping all other defined ones within a route""" return self.where( requires=tuple( set(self.route.get("requires", ())).difference( requirements if type(requirements) in (list, tuple) else (requirements,) ) ) )
map_params
def map_params( self, **map_params )
Map interface specific params to an internal name representation
View Source
def map_params(self, **map_params): """Map interface specific params to an internal name representation""" return self.where(map_params=map_params)
on_invalid
def on_invalid( self, function, **overrides )
Sets a function to use to transform data on validation errors.
Defaults to the transform function if one is set to ensure no special
handling occurs for invalid data set to False.
View Source
def on_invalid(self, function, **overrides): """Sets a function to use to transform data on validation errors. Defaults to the transform function if one is set to ensure no special handling occurs for invalid data set to `False`. """ return self.where(on_invalid=function, **overrides)
output
def output( self, formatter, **overrides )
Sets the output formatter that should be used to render this route
View Source
def output(self, formatter, **overrides): """Sets the output formatter that should be used to render this route""" return self.where(output=formatter, **overrides)
output_invalid
def output_invalid( self, output_handler, **overrides )
Sets an output handler to be used when handler validation fails.
Defaults to the output formatter set globally for the route.
View Source
def output_invalid(self, output_handler, **overrides): """Sets an output handler to be used when handler validation fails. Defaults to the output formatter set globally for the route. """ return self.where(output_invalid=output_handler, **overrides)
parameters
def parameters( self, parameters, **overrides )
Sets the custom parameters that will be used instead of those found introspecting the decorated function
View Source
def parameters(self, parameters, **overrides): """Sets the custom parameters that will be used instead of those found introspecting the decorated function""" return self.where(parameters=parameters, **overrides)
parse_body
def parse_body( self, automatic=True, **overrides )
Tells hug to automatically parse the input body if it matches a registered input format
View Source
def parse_body(self, automatic=True, **overrides): """Tells hug to automatically parse the input body if it matches a registered input format""" return self.where(parse_body=automatic, **overrides)
raise_on_invalid
def raise_on_invalid( self, setting=True, **overrides )
Sets the route to raise validation errors instead of catching them
View Source
def raise_on_invalid(self, setting=True, **overrides): """Sets the route to raise validation errors instead of catching them""" return self.where(raise_on_invalid=setting, **overrides)
requires
def requires( self, requirements, **overrides )
Adds additional requirements to the specified route
View Source
def requires(self, requirements, **overrides): """Adds additional requirements to the specified route""" return self.where( requires=tuple(self.route.get("requires", ())) + tuple(requirements), **overrides )
response_headers
def response_headers( self, headers, **overrides )
Sets the response headers automatically injected by the router
View Source
def response_headers(self, headers, **overrides): """Sets the response headers automatically injected by the router""" return self.where(response_headers=headers, **overrides)
set_status
def set_status( self, status, **overrides )
Sets the status that will be returned by default
View Source
def set_status(self, status, **overrides): """Sets the status that will be returned by default""" return self.where(status=status, **overrides)
transform
def transform( self, function, **overrides )
Sets the function that should be used to transform the returned Python structure into something serializable by specified output format
View Source
def transform(self, function, **overrides): """Sets the function that should be used to transform the returned Python structure into something serializable by specified output format """ return self.where(transform=function, **overrides)
validate
def validate( self, validation_function, **overrides )
Sets the secondary validation fucntion to use for this handler
View Source
def validate(self, validation_function, **overrides): """Sets the secondary validation fucntion to use for this handler""" return self.where(validate=validation_function, **overrides)
versions
def versions( self, supported, **overrides )
Sets the versions that this route should be compatiable with
View Source
def versions(self, supported, **overrides): """Sets the versions that this route should be compatiable with""" return self.where(versions=supported, **overrides)
where
def where( self, **overrides )
Creates a new route, based on the current route, with the specified overrided values
View Source
def where(self, **overrides): """Creates a new route, based on the current route, with the specified overrided values""" route_data = self.route.copy() route_data.update(overrides) return self.__class__(**route_data)
URLRouter
class URLRouter( urls=None, accept=('CONNECT', 'DELETE', 'GET', 'HEAD', 'OPTIONS', 'PATCH', 'POST', 'PUT', 'TRACE'), output=None, examples=(), versions=<built-in function any>, suffixes=(), prefixes=(), response_headers=None, parse_body=True, **kwargs )
Provides a chainable router that can be used to route a URL to a Python function
View Source
class URLRouter(HTTPRouter): """Provides a chainable router that can be used to route a URL to a Python function""" __slots__ = () def __init__( self, urls=None, accept=HTTP_METHODS, output=None, examples=(), versions=any, suffixes=(), prefixes=(), response_headers=None, parse_body=True, **kwargs ): super().__init__( output=output, versions=versions, parse_body=parse_body, response_headers=response_headers, **kwargs ) if urls is not None: self.route["urls"] = (urls,) if isinstance(urls, str) else urls if accept: self.route["accept"] = (accept,) if isinstance(accept, str) else accept if examples: self.route["examples"] = (examples,) if isinstance(examples, str) else examples if suffixes: self.route["suffixes"] = (suffixes,) if isinstance(suffixes, str) else suffixes if prefixes: self.route["prefixes"] = (prefixes,) if isinstance(prefixes, str) else prefixes def __call__(self, api_function): api = self.route.get("api", hug.api.from_object(api_function)) api.http.routes.setdefault(api.http.base_url, OrderedDict()) (interface, callable_method) = self._create_interface(api, api_function) use_examples = self.route.get("examples", ()) if not interface.required and not use_examples: use_examples = (True,) for base_url in self.route.get("urls", ("/{0}".format(api_function.__name__),)): expose = [base_url] for suffix in self.route.get("suffixes", ()): if suffix.startswith("/"): expose.append(os.path.join(base_url, suffix.lstrip("/"))) else: expose.append(base_url + suffix) for prefix in self.route.get("prefixes", ()): expose.append(prefix + base_url) for url in expose: handlers = api.http.routes[api.http.base_url].setdefault(url, {}) for method in self.route.get("accept", ()): version_mapping = handlers.setdefault(method.upper(), {}) for version in self.route.get("versions", (None,)): version_mapping[version] = interface api.http.versioned.setdefault(version, {})[ callable_method.__name__ ] = callable_method interface.examples = use_examples return callable_method def urls(self, *urls, **overrides): """Sets the URLs that will map to this API call""" return self.where(urls=urls, **overrides) def accept(self, *accept, **overrides): """Sets a list of HTTP methods this router should accept""" return self.where(accept=accept, **overrides) def get(self, urls=None, **overrides): """Sets the acceptable HTTP method to a GET""" if urls is not None: overrides["urls"] = urls return self.where(accept="GET", **overrides) def delete(self, urls=None, **overrides): """Sets the acceptable HTTP method to DELETE""" if urls is not None: overrides["urls"] = urls return self.where(accept="DELETE", **overrides) def post(self, urls=None, **overrides): """Sets the acceptable HTTP method to POST""" if urls is not None: overrides["urls"] = urls return self.where(accept="POST", **overrides) def put(self, urls=None, **overrides): """Sets the acceptable HTTP method to PUT""" if urls is not None: overrides["urls"] = urls return self.where(accept="PUT", **overrides) def trace(self, urls=None, **overrides): """Sets the acceptable HTTP method to TRACE""" if urls is not None: overrides["urls"] = urls return self.where(accept="TRACE", **overrides) def patch(self, urls=None, **overrides): """Sets the acceptable HTTP method to PATCH""" if urls is not None: overrides["urls"] = urls return self.where(accept="PATCH", **overrides) def options(self, urls=None, **overrides): """Sets the acceptable HTTP method to OPTIONS""" if urls is not None: overrides["urls"] = urls return self.where(accept="OPTIONS", **overrides) def head(self, urls=None, **overrides): """Sets the acceptable HTTP method to HEAD""" if urls is not None: overrides["urls"] = urls return self.where(accept="HEAD", **overrides) def connect(self, urls=None, **overrides): """Sets the acceptable HTTP method to CONNECT""" if urls is not None: overrides["urls"] = urls return self.where(accept="CONNECT", **overrides) def call(self, **overrides): """Sets the acceptable HTTP method to all known""" return self.where(accept=HTTP_METHODS, **overrides) def http(self, **overrides): """Sets the acceptable HTTP method to all known""" return self.where(accept=HTTP_METHODS, **overrides) def get_post(self, **overrides): """Exposes a Python method externally under both the HTTP POST and GET methods""" return self.where(accept=("GET", "POST"), **overrides) def put_post(self, **overrides): """Exposes a Python method externally under both the HTTP POST and PUT methods""" return self.where(accept=("PUT", "POST"), **overrides) def examples(self, *examples, **overrides): """Sets the examples that the route should use""" return self.where(examples=examples, **overrides) def suffixes(self, *suffixes, **overrides): """Sets the suffixes supported by the route""" return self.where(suffixes=suffixes, **overrides) def prefixes(self, *prefixes, **overrides): """Sets the prefixes supported by the route""" return self.where(prefixes=prefixes, **overrides) def where(self, **overrides): if "urls" in overrides: existing_urls = self.route.get("urls", ()) use_urls = [] for url in ( (overrides["urls"],) if isinstance(overrides["urls"], str) else overrides["urls"] ): if url.startswith("/") or not existing_urls: use_urls.append(url) else: for existing in existing_urls: use_urls.append(urljoin(existing.rstrip("/") + "/", url)) overrides["urls"] = tuple(use_urls) return super().where(**overrides)
Ancestors (in MRO)
- hug.routing.HTTPRouter
- hug.routing.InternalValidation
- hug.routing.Router
Descendants
- hug.route.Object
Class variables
route
Instance variables
route
Methods
accept
def accept( self, *accept, **overrides )
Sets a list of HTTP methods this router should accept
View Source
def accept(self, *accept, **overrides): """Sets a list of HTTP methods this router should accept""" return self.where(accept=accept, **overrides)
add_response_headers
def add_response_headers( self, headers, **overrides )
Adds the specified response headers while keeping existing ones in-tact
View Source
def add_response_headers(self, headers, **overrides): """Adds the specified response headers while keeping existing ones in-tact""" response_headers = self.route.get("response_headers", {}).copy() response_headers.update(headers) return self.where(response_headers=response_headers, **overrides)
allow_origins
def allow_origins( self, *origins, methods=None, max_age=None, credentials=None, headers=None, **overrides )
Convenience method for quickly allowing other resources to access this one
View Source
def allow_origins( self, *origins, methods=None, max_age=None, credentials=None, headers=None, **overrides ): """Convenience method for quickly allowing other resources to access this one""" response_headers = {} if origins: @hug.response_middleware(api=self.route.get("api", None)) def process_data(request, response, resource): if "ORIGIN" in request.headers: origin = request.headers["ORIGIN"] if origin in origins: response.set_header("Access-Control-Allow-Origin", origin) else: response_headers["Access-Control-Allow-Origin"] = "*" if methods: response_headers["Access-Control-Allow-Methods"] = ", ".join(methods) if max_age: response_headers["Access-Control-Max-Age"] = max_age if credentials: response_headers["Access-Control-Allow-Credentials"] = str(credentials).lower() if headers: response_headers["Access-Control-Allow-Headers"] = headers return self.add_response_headers(response_headers, **overrides)
api
def api( self, api, **overrides )
Sets the API that should contain this route
View Source
def api(self, api, **overrides): """Sets the API that should contain this route""" return self.where(api=api, **overrides)
cache
def cache( self, private=False, max_age=31536000, s_maxage=None, no_cache=False, no_store=False, must_revalidate=False, **overrides )
Convenience method for quickly adding cache header to route
View Source
def cache( self, private=False, max_age=31536000, s_maxage=None, no_cache=False, no_store=False, must_revalidate=False, **overrides ): """Convenience method for quickly adding cache header to route""" parts = ( "private" if private else "public", "max-age={0}".format(max_age), "s-maxage={0}".format(s_maxage) if s_maxage is not None else None, no_cache and "no-cache", no_store and "no-store", must_revalidate and "must-revalidate", ) return self.add_response_headers( {"cache-control": ", ".join(filter(bool, parts))}, **overrides )
call
def call( self, **overrides )
Sets the acceptable HTTP method to all known
View Source
def call(self, **overrides): """Sets the acceptable HTTP method to all known""" return self.where(accept=HTTP_METHODS, **overrides)
connect
def connect( self, urls=None, **overrides )
Sets the acceptable HTTP method to CONNECT
View Source
def connect(self, urls=None, **overrides): """Sets the acceptable HTTP method to CONNECT""" if urls is not None: overrides["urls"] = urls return self.where(accept="CONNECT", **overrides)
defaults
def defaults( self, defaults, **overrides )
Sets the custom defaults that will be used for custom parameters
View Source
def defaults(self, defaults, **overrides): """Sets the custom defaults that will be used for custom parameters""" return self.where(defaults=defaults, **overrides)
delete
def delete( self, urls=None, **overrides )
Sets the acceptable HTTP method to DELETE
View Source
def delete(self, urls=None, **overrides): """Sets the acceptable HTTP method to DELETE""" if urls is not None: overrides["urls"] = urls return self.where(accept="DELETE", **overrides)
doesnt_require
def doesnt_require( self, requirements, **overrides )
Removes individual requirements while keeping all other defined ones within a route
View Source
def doesnt_require(self, requirements, **overrides): """Removes individual requirements while keeping all other defined ones within a route""" return self.where( requires=tuple( set(self.route.get("requires", ())).difference( requirements if type(requirements) in (list, tuple) else (requirements,) ) ) )
examples
def examples( self, *examples, **overrides )
Sets the examples that the route should use
View Source
def examples(self, *examples, **overrides): """Sets the examples that the route should use""" return self.where(examples=examples, **overrides)
get
def get( self, urls=None, **overrides )
Sets the acceptable HTTP method to a GET
View Source
def get(self, urls=None, **overrides): """Sets the acceptable HTTP method to a GET""" if urls is not None: overrides["urls"] = urls return self.where(accept="GET", **overrides)
get_post
def get_post( self, **overrides )
Exposes a Python method externally under both the HTTP POST and GET methods
View Source
def get_post(self, **overrides): """Exposes a Python method externally under both the HTTP POST and GET methods""" return self.where(accept=("GET", "POST"), **overrides)
head
def head( self, urls=None, **overrides )
Sets the acceptable HTTP method to HEAD
View Source
def head(self, urls=None, **overrides): """Sets the acceptable HTTP method to HEAD""" if urls is not None: overrides["urls"] = urls return self.where(accept="HEAD", **overrides)
http
def http( self, **overrides )
Sets the acceptable HTTP method to all known
View Source
def http(self, **overrides): """Sets the acceptable HTTP method to all known""" return self.where(accept=HTTP_METHODS, **overrides)
map_params
def map_params( self, **map_params )
Map interface specific params to an internal name representation
View Source
def map_params(self, **map_params): """Map interface specific params to an internal name representation""" return self.where(map_params=map_params)
on_invalid
def on_invalid( self, function, **overrides )
Sets a function to use to transform data on validation errors.
Defaults to the transform function if one is set to ensure no special
handling occurs for invalid data set to False.
View Source
def on_invalid(self, function, **overrides): """Sets a function to use to transform data on validation errors. Defaults to the transform function if one is set to ensure no special handling occurs for invalid data set to `False`. """ return self.where(on_invalid=function, **overrides)
options
def options( self, urls=None, **overrides )
Sets the acceptable HTTP method to OPTIONS
View Source
def options(self, urls=None, **overrides): """Sets the acceptable HTTP method to OPTIONS""" if urls is not None: overrides["urls"] = urls return self.where(accept="OPTIONS", **overrides)
output
def output( self, formatter, **overrides )
Sets the output formatter that should be used to render this route
View Source
def output(self, formatter, **overrides): """Sets the output formatter that should be used to render this route""" return self.where(output=formatter, **overrides)
output_invalid
def output_invalid( self, output_handler, **overrides )
Sets an output handler to be used when handler validation fails.
Defaults to the output formatter set globally for the route.
View Source
def output_invalid(self, output_handler, **overrides): """Sets an output handler to be used when handler validation fails. Defaults to the output formatter set globally for the route. """ return self.where(output_invalid=output_handler, **overrides)
parameters
def parameters( self, parameters, **overrides )
Sets the custom parameters that will be used instead of those found introspecting the decorated function
View Source
def parameters(self, parameters, **overrides): """Sets the custom parameters that will be used instead of those found introspecting the decorated function""" return self.where(parameters=parameters, **overrides)
parse_body
def parse_body( self, automatic=True, **overrides )
Tells hug to automatically parse the input body if it matches a registered input format
View Source
def parse_body(self, automatic=True, **overrides): """Tells hug to automatically parse the input body if it matches a registered input format""" return self.where(parse_body=automatic, **overrides)
patch
def patch( self, urls=None, **overrides )
Sets the acceptable HTTP method to PATCH
View Source
def patch(self, urls=None, **overrides): """Sets the acceptable HTTP method to PATCH""" if urls is not None: overrides["urls"] = urls return self.where(accept="PATCH", **overrides)
post
def post( self, urls=None, **overrides )
Sets the acceptable HTTP method to POST
View Source
def post(self, urls=None, **overrides): """Sets the acceptable HTTP method to POST""" if urls is not None: overrides["urls"] = urls return self.where(accept="POST", **overrides)
prefixes
def prefixes( self, *prefixes, **overrides )
Sets the prefixes supported by the route
View Source
def prefixes(self, *prefixes, **overrides): """Sets the prefixes supported by the route""" return self.where(prefixes=prefixes, **overrides)
put
def put( self, urls=None, **overrides )
Sets the acceptable HTTP method to PUT
View Source
def put(self, urls=None, **overrides): """Sets the acceptable HTTP method to PUT""" if urls is not None: overrides["urls"] = urls return self.where(accept="PUT", **overrides)
put_post
def put_post( self, **overrides )
Exposes a Python method externally under both the HTTP POST and PUT methods
View Source
def put_post(self, **overrides): """Exposes a Python method externally under both the HTTP POST and PUT methods""" return self.where(accept=("PUT", "POST"), **overrides)
raise_on_invalid
def raise_on_invalid( self, setting=True, **overrides )
Sets the route to raise validation errors instead of catching them
View Source
def raise_on_invalid(self, setting=True, **overrides): """Sets the route to raise validation errors instead of catching them""" return self.where(raise_on_invalid=setting, **overrides)
requires
def requires( self, requirements, **overrides )
Adds additional requirements to the specified route
View Source
def requires(self, requirements, **overrides): """Adds additional requirements to the specified route""" return self.where( requires=tuple(self.route.get("requires", ())) + tuple(requirements), **overrides )
response_headers
def response_headers( self, headers, **overrides )
Sets the response headers automatically injected by the router
View Source
def response_headers(self, headers, **overrides): """Sets the response headers automatically injected by the router""" return self.where(response_headers=headers, **overrides)
set_status
def set_status( self, status, **overrides )
Sets the status that will be returned by default
View Source
def set_status(self, status, **overrides): """Sets the status that will be returned by default""" return self.where(status=status, **overrides)
suffixes
def suffixes( self, *suffixes, **overrides )
Sets the suffixes supported by the route
View Source
def suffixes(self, *suffixes, **overrides): """Sets the suffixes supported by the route""" return self.where(suffixes=suffixes, **overrides)
trace
def trace( self, urls=None, **overrides )
Sets the acceptable HTTP method to TRACE
View Source
def trace(self, urls=None, **overrides): """Sets the acceptable HTTP method to TRACE""" if urls is not None: overrides["urls"] = urls return self.where(accept="TRACE", **overrides)
transform
def transform( self, function, **overrides )
Sets the function that should be used to transform the returned Python structure into something serializable by specified output format
View Source
def transform(self, function, **overrides): """Sets the function that should be used to transform the returned Python structure into something serializable by specified output format """ return self.where(transform=function, **overrides)
urls
def urls( self, *urls, **overrides )
Sets the URLs that will map to this API call
View Source
def urls(self, *urls, **overrides): """Sets the URLs that will map to this API call""" return self.where(urls=urls, **overrides)
validate
def validate( self, validation_function, **overrides )
Sets the secondary validation fucntion to use for this handler
View Source
def validate(self, validation_function, **overrides): """Sets the secondary validation fucntion to use for this handler""" return self.where(validate=validation_function, **overrides)
versions
def versions( self, supported, **overrides )
Sets the versions that this route should be compatiable with
View Source
def versions(self, supported, **overrides): """Sets the versions that this route should be compatiable with""" return self.where(versions=supported, **overrides)
where
def where( self, **overrides )
Creates a new route, based on the current route, with the specified overrided values
View Source
def where(self, **overrides): if "urls" in overrides: existing_urls = self.route.get("urls", ()) use_urls = [] for url in ( (overrides["urls"],) if isinstance(overrides["urls"], str) else overrides["urls"] ): if url.startswith("/") or not existing_urls: use_urls.append(url) else: for existing in existing_urls: use_urls.append(urljoin(existing.rstrip("/") + "/", url)) overrides["urls"] = tuple(use_urls) return super().where(**overrides)