Vasanko/sanic
2
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge pull request #12 from channelcat/master

Browse Source 
merge upstream master branch
This commit is contained in:
7 2017-09-08 17:39:50 -07:00 committed by GitHub
commit c2a3e42a53
18 changed files with 927 additions and 121 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
LICENSE
View File

@ -1,6 +1,6 @@
MIT License
Copyright (c) [year] [fullname]
Copyright (c) 2016-present Channel Cat
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
@ -18,4 +18,4 @@ 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.
SOFTWARE.

10
docs/sanic/config.md
View File

@ -29,9 +29,15 @@ In general the convention is to only have UPPERCASE configuration parameters. Th
There are several ways how to load configuration.
### From environment variables.
### From Environment Variables
Any variables defined with the `SANIC_` prefix will be applied to the sanic config. For example, setting `SANIC_REQUEST_TIMEOUT` will be loaded by the application automatically. You can pass the `load_env` boolean to the Sanic constructor to override that:
Any variables defined with the `SANIC_` prefix will be applied to the sanic config. For example, setting `SANIC_REQUEST_TIMEOUT` will be loaded by the application automatically and fed into the `REQUEST_TIMEOUT` config variable. You can pass a different prefix to Sanic:
```python
app = Sanic(load_env='MYAPP_')
```
Then the above variable would be `MYAPP_REQUEST_TIMEOUT`. If you want to disable loading from environment variables you can set it to `False` instead:
```python
app = Sanic(load_env=False)

2
docs/sanic/request_data.md
View File

@ -71,6 +71,8 @@ The following variables are accessible as properties on `Request` objects:
return text("You are trying to create a user with the following POST: %s" % request.body)
```
- `headers` (dict) - A case-insensitive dictionary that contains the request headers.
- `ip` (str) - IP address of the requester.
- `app` - a reference to the Sanic application object that is handling this request. This is useful when inside blueprints or other handlers in modules that do not have access to the global `app` object.

87
docs/sanic/routing.md
View File

@ -214,3 +214,90 @@ and `recv` methods to send and receive data respectively.
WebSocket support requires the [websockets](https://github.com/aaugustin/websockets)
package by Aymeric Augustin.
## About `strict_slashes`
You can make `routes` strict to trailing slash or not, it's configurable.
```python
# provide default strict_slashes value for all routes
app = Sanic('test_route_strict_slash', strict_slashes=True)
# you can also overwrite strict_slashes value for specific route
@app.get('/get', strict_slashes=False)
def handler(request):
return text('OK')
# It also works for blueprints
bp = Blueprint('test_bp_strict_slash', strict_slashes=True)
@bp.get('/bp/get', strict_slashes=False)
def handler(request):
return text('OK')
app.blueprint(bp)
```
## User defined route name
You can pass `name` to change the route name to avoid using the default name (`handler.__name__`).
```python
app = Sanic('test_named_route')
@app.get('/get', name='get_handler')
def handler(request):
return text('OK')
# then you need use `app.url_for('get_handler')`
# instead of # `app.url_for('handler')`
# It also works for blueprints
bp = Blueprint('test_named_bp')
@bp.get('/bp/get', name='get_handler')
def handler(request):
return text('OK')
app.blueprint(bp)
# then you need use `app.url_for('test_named_bp.get_handler')`
# instead of `app.url_for('test_named_bp.handler')`
# different names can be used for same url with different methods
@app.get('/test', name='route_test')
def handler(request):
return text('OK')
@app.post('/test', name='route_post')
def handler2(request):
return text('OK POST')
@app.put('/test', name='route_put')
def handler3(request):
return text('OK PUT')
# below url are the same, you can use any of them
# '/test'
app.url_for('route_test')
# app.url_for('route_post')
# app.url_for('route_put')
# for same handler name with different methods
# you need specify the name (it's url_for issue)
@app.get('/get')
def handler(request):
return text('OK')
@app.post('/post', name='post_handler')
def handler(request):
return text('OK')
# then
# app.url_for('handler') == '/get'
# app.url_for('post_handler') == '/post'
```

17
examples/add_task_sanic.py Normal file
View File

@ -0,0 +1,17 @@
# -*- coding: utf-8 -*-
import asyncio
from sanic import Sanic
app = Sanic()
async def notify_server_started_after_five_seconds():
await asyncio.sleep(5)
print('Server successfully started!')
app.add_task(notify_server_started_after_five_seconds())
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8000)

42
examples/authorized_sanic.py Normal file
View File

@ -0,0 +1,42 @@
# -*- coding: utf-8 -*-
from sanic import Sanic
from functools import wraps
from sanic.response import json
app = Sanic()
def check_request_for_authorization_status(request):
# Note: Define your check, for instance cookie, session.
flag = True
return flag
def authorized():
def decorator(f):
@wraps(f)
async def decorated_function(request, *args, **kwargs):
# run some method that checks the request
# for the client's authorization status
is_authorized = check_request_for_authorization_status(request)
if is_authorized:
# the user is authorized.
# run the handler method and return the response
response = await f(request, *args, **kwargs)
return response
else:
# the user is not authorized.
return json({'status': 'not_authorized'}, 403)
return decorated_function
return decorator
@app.route("/")
@authorized()
async def test(request):
return json({'status': 'authorized'})
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8000)

86
examples/log_request_id.py Normal file
View File

@ -0,0 +1,86 @@
'''
Based on example from https://github.com/Skyscanner/aiotask-context
and `examples/{override_logging,run_async}.py`.
Needs https://github.com/Skyscanner/aiotask-context/tree/52efbc21e2e1def2d52abb9a8e951f3ce5e6f690 or newer
$ pip install git+https://github.com/Skyscanner/aiotask-context.git
'''
import asyncio
import uuid
import logging
from signal import signal, SIGINT
from sanic import Sanic
from sanic import response
import uvloop
import aiotask_context as context
log = logging.getLogger(__name__)
class RequestIdFilter(logging.Filter):
def filter(self, record):
record.request_id = context.get('X-Request-ID')
return True
LOG_SETTINGS = {
'version': 1,
'disable_existing_loggers': False,
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'level': 'DEBUG',
'formatter': 'default',
'filters': ['requestid'],
},
},
'filters': {
'requestid': {
'()': RequestIdFilter,
},
},
'formatters': {
'default': {
'format': '%(asctime)s %(levelname)s %(name)s:%(lineno)d %(request_id)s | %(message)s',
},
},
'loggers': {
'': {
'level': 'DEBUG',
'handlers': ['console'],
'propagate': True
},
}
}
app = Sanic(__name__, log_config=LOG_SETTINGS)
@app.middleware('request')
async def set_request_id(request):
request_id = request.headers.get('X-Request-ID') or str(uuid.uuid4())
context.set("X-Request-ID", request_id)
@app.route("/")
async def test(request):
log.debug('X-Request-ID: %s', context.get('X-Request-ID'))
log.info('Hello from test!')
return response.json({"test": True})
if __name__ == '__main__':
asyncio.set_event_loop(uvloop.new_event_loop())
server = app.create_server(host="0.0.0.0", port=8000)
loop = asyncio.get_event_loop()
loop.set_task_factory(context.task_factory)
task = asyncio.ensure_future(server)
try:
loop.run_forever()
except:
loop.stop()

2
examples/try_everything.py
View File

@ -18,7 +18,7 @@ def test_sync(request):
return response.json({"test": True})
@app.route("/dynamic/<name>/<id:int>")
@app.route("/dynamic/<name>/<i:int>")
def test_params(request, name, i):
return response.text("yeehaww {} {}".format(name, i))

87
sanic/app.py
View File

@ -28,7 +28,7 @@ class Sanic:
def __init__(self, name=None, router=None, error_handler=None,
load_env=True, request_class=None,
log_config=LOGGING):
log_config=LOGGING, strict_slashes=False):
if log_config:
logging.config.dictConfig(log_config)
# Only set up a default log handler if the
@ -58,6 +58,7 @@ class Sanic:
self._blueprint_order = []
self.debug = None
self.sock = None
self.strict_slashes = strict_slashes
self.listeners = defaultdict(list)
self.is_running = False
self.is_request_stream = False
@ -111,7 +112,7 @@ class Sanic:
# Decorator
def route(self, uri, methods=frozenset({'GET'}), host=None,
strict_slashes=False, stream=False, version=None):
strict_slashes=None, stream=False, version=None, name=None):
"""Decorate a function to be registered as a route
:param uri: path of the URL
@ -119,6 +120,8 @@ class Sanic:
:param host:
:param strict_slashes:
:param stream:
:param version:
:param name: user defined route name for url_for
:return: decorated function
"""
@ -130,53 +133,64 @@ class Sanic:
if stream:
self.is_request_stream = True
if strict_slashes is None:
strict_slashes = self.strict_slashes
def response(handler):
if stream:
handler.is_stream = stream
self.router.add(uri=uri, methods=methods, handler=handler,
host=host, strict_slashes=strict_slashes,
version=version)
version=version, name=name)
return handler
return response
# Shorthand method decorators
def get(self, uri, host=None, strict_slashes=False, version=None):
def get(self, uri, host=None, strict_slashes=None, version=None,
name=None):
return self.route(uri, methods=frozenset({"GET"}), host=host,
strict_slashes=strict_slashes, version=version)
strict_slashes=strict_slashes, version=version,
name=name)
def post(self, uri, host=None, strict_slashes=False, stream=False,
version=None):
def post(self, uri, host=None, strict_slashes=None, stream=False,
version=None, name=None):
return self.route(uri, methods=frozenset({"POST"}), host=host,
strict_slashes=strict_slashes, stream=stream,
version=version)
version=version, name=name)
def put(self, uri, host=None, strict_slashes=False, stream=False,
version=None):
def put(self, uri, host=None, strict_slashes=None, stream=False,
version=None, name=None):
return self.route(uri, methods=frozenset({"PUT"}), host=host,
strict_slashes=strict_slashes, stream=stream,
version=version)
version=version, name=name)
def head(self, uri, host=None, strict_slashes=False, version=None):
def head(self, uri, host=None, strict_slashes=None, version=None,
name=None):
return self.route(uri, methods=frozenset({"HEAD"}), host=host,
strict_slashes=strict_slashes, version=version)
strict_slashes=strict_slashes, version=version,
name=name)
def options(self, uri, host=None, strict_slashes=False, version=None):
def options(self, uri, host=None, strict_slashes=None, version=None,
name=None):
return self.route(uri, methods=frozenset({"OPTIONS"}), host=host,
strict_slashes=strict_slashes, version=version)
strict_slashes=strict_slashes, version=version,
name=name)
def patch(self, uri, host=None, strict_slashes=False, stream=False,
version=None):
def patch(self, uri, host=None, strict_slashes=None, stream=False,
version=None, name=None):
return self.route(uri, methods=frozenset({"PATCH"}), host=host,
strict_slashes=strict_slashes, stream=stream,
version=version)
version=version, name=name)
def delete(self, uri, host=None, strict_slashes=False, version=None):
def delete(self, uri, host=None, strict_slashes=None, version=None,
name=None):
return self.route(uri, methods=frozenset({"DELETE"}), host=host,
strict_slashes=strict_slashes, version=version)
strict_slashes=strict_slashes, version=version,
name=name)
def add_route(self, handler, uri, methods=frozenset({'GET'}), host=None,
strict_slashes=False, version=None):
strict_slashes=None, version=None, name=None):
"""A helper method to register class instance or
functions as a handler to the application url
routes.
@ -186,6 +200,9 @@ class Sanic:
:param methods: list or tuple of methods allowed, these are overridden
if using a HTTPMethodView
:param host:
:param strict_slashes:
:param version:
:param name: user defined route name for url_for
:return: function or class instance
"""
stream = False
@ -208,14 +225,17 @@ class Sanic:
stream = True
break
if strict_slashes is None:
strict_slashes = self.strict_slashes
self.route(uri=uri, methods=methods, host=host,
strict_slashes=strict_slashes, stream=stream,
version=version)(handler)
version=version, name=name)(handler)
return handler
# Decorator
def websocket(self, uri, host=None, strict_slashes=False,
subprotocols=None):
def websocket(self, uri, host=None, strict_slashes=None,
subprotocols=None, name=None):
"""Decorate a function to be registered as a websocket route
:param uri: path of the URL
:param subprotocols: optional list of strings with the supported
@ -230,6 +250,9 @@ class Sanic:
if not uri.startswith('/'):
uri = '/' + uri
if strict_slashes is None:
strict_slashes = self.strict_slashes
def response(handler):
async def websocket_handler(request, *args, **kwargs):
request.app = self
@ -255,16 +278,19 @@ class Sanic:
self.router.add(uri=uri, handler=websocket_handler,
methods=frozenset({'GET'}), host=host,
strict_slashes=strict_slashes)
strict_slashes=strict_slashes, name=name)
return handler
return response
def add_websocket_route(self, handler, uri, host=None,
strict_slashes=False):
strict_slashes=None, name=None):
"""A helper method to register a function as a websocket route."""
return self.websocket(uri, host=host,
strict_slashes=strict_slashes)(handler)
if strict_slashes is None:
strict_slashes = self.strict_slashes
return self.websocket(uri, host=host, strict_slashes=strict_slashes,
name=name)(handler)
def enable_websocket(self, enable=True):
"""Enable or disable the support for websocket.
@ -387,9 +413,8 @@ class Sanic:
uri, route = self.router.find_route_by_view_name(view_name)
if not uri or not route:
raise URLBuildError(
'Endpoint with name `{}` was not found'.format(
view_name))
raise URLBuildError('Endpoint with name `{}` was not found'.format(
view_name))
if uri != '/' and uri.endswith('/'):
uri = uri[:-1]

111
sanic/blueprints.py
View File

@ -5,7 +5,7 @@ from sanic.views import CompositionView
FutureRoute = namedtuple('Route',
['handler', 'uri', 'methods', 'host',
'strict_slashes', 'stream', 'version'])
'strict_slashes', 'stream', 'version', 'name'])
FutureListener = namedtuple('Listener', ['handler', 'uri', 'methods', 'host'])
FutureMiddleware = namedtuple('Route', ['middleware', 'args', 'kwargs'])
FutureException = namedtuple('Route', ['handler', 'args', 'kwargs'])
@ -14,11 +14,16 @@ FutureStatic = namedtuple('Route',
class Blueprint:
def __init__(self, name, url_prefix=None, host=None, version=None):
def __init__(self, name,
url_prefix=None,
host=None, version=None,
strict_slashes=False):
"""Create a new blueprint
:param name: unique name of the blueprint
:param url_prefix: URL to be prefixed before all route URLs
:param strict_slashes: strict to trailing slash
"""
self.name = name
self.url_prefix = url_prefix
@ -31,6 +36,7 @@ class Blueprint:
self.middlewares = []
self.statics = []
self.version = version
self.strict_slashes = strict_slashes
def register(self, app, options):
"""Register the blueprint to the sanic app."""
@ -47,14 +53,14 @@ class Blueprint:
version = future.version or self.version
app.route(
uri=uri[1:] if uri.startswith('//') else uri,
methods=future.methods,
host=future.host or self.host,
strict_slashes=future.strict_slashes,
stream=future.stream,
version=version
)(future.handler)
app.route(uri=uri[1:] if uri.startswith('//') else uri,
methods=future.methods,
host=future.host or self.host,
strict_slashes=future.strict_slashes,
stream=future.stream,
version=version,
name=future.name,
)(future.handler)
for future in self.websocket_routes:
# attach the blueprint name to the handler so that it can be
@ -62,11 +68,11 @@ class Blueprint:
future.handler.__blueprintname__ = self.name
# Prepend the blueprint URI prefix if available
uri = url_prefix + future.uri if url_prefix else future.uri
app.websocket(
uri=uri,
host=future.host or self.host,
strict_slashes=future.strict_slashes
)(future.handler)
app.websocket(uri=uri,
host=future.host or self.host,
strict_slashes=future.strict_slashes,
name=future.name,
)(future.handler)
# Middleware
for future in self.middlewares:
@ -94,27 +100,35 @@ class Blueprint:
app.listener(event)(listener)
def route(self, uri, methods=frozenset({'GET'}), host=None,
strict_slashes=False, stream=False, version=None):
strict_slashes=None, stream=False, version=None, name=None):
"""Create a blueprint route from a decorated function.
:param uri: endpoint at which the route will be accessible.
:param methods: list of acceptable HTTP methods.
"""
if strict_slashes is None:
strict_slashes = self.strict_slashes
def decorator(handler):
route = FutureRoute(
handler, uri, methods, host, strict_slashes, stream, version)
handler, uri, methods, host, strict_slashes, stream, version,
name)
self.routes.append(route)
return handler
return decorator
def add_route(self, handler, uri, methods=frozenset({'GET'}), host=None,
strict_slashes=False, version=None):
strict_slashes=None, version=None, name=None):
"""Create a blueprint route from a function.
:param handler: function for handling uri requests. Accepts function,
or class instance with a view_class method.
:param uri: endpoint at which the route will be accessible.
:param methods: list of acceptable HTTP methods.
:param host:
:param strict_slashes:
:param version:
:param name: user defined route name for url_for
:return: function or class instance
"""
# Handle HTTPMethodView differently
@ -125,27 +139,36 @@ class Blueprint:
if getattr(handler.view_class, method.lower(), None):
methods.add(method)
if strict_slashes is None:
strict_slashes = self.strict_slashes
# handle composition view differently
if isinstance(handler, CompositionView):
methods = handler.handlers.keys()
self.route(uri=uri, methods=methods, host=host,
strict_slashes=strict_slashes, version=version)(handler)
strict_slashes=strict_slashes, version=version,
name=name)(handler)
return handler
def websocket(self, uri, host=None, strict_slashes=False, version=None):
def websocket(self, uri, host=None, strict_slashes=None, version=None,
name=None):
"""Create a blueprint websocket route from a decorated function.
:param uri: endpoint at which the route will be accessible.
"""
if strict_slashes is None:
strict_slashes = self.strict_slashes
def decorator(handler):
route = FutureRoute(handler, uri, [], host, strict_slashes,
False, version)
False, version, name)
self.websocket_routes.append(route)
return handler
return decorator
def add_websocket_route(self, handler, uri, host=None, version=None):
def add_websocket_route(self, handler, uri, host=None, version=None,
name=None):
"""Create a blueprint websocket route from a function.
:param handler: function for handling uri requests. Accepts function,
@ -153,7 +176,7 @@ class Blueprint:
:param uri: endpoint at which the route will be accessible.
:return: function or class instance
"""
self.websocket(uri=uri, host=host, version=version)(handler)
self.websocket(uri=uri, host=host, version=version, name=name)(handler)
return handler
def listener(self, event):
@ -199,36 +222,44 @@ class Blueprint:
self.statics.append(static)
# Shorthand method decorators
def get(self, uri, host=None, strict_slashes=False, version=None):
def get(self, uri, host=None, strict_slashes=None, version=None,
name=None):
return self.route(uri, methods=["GET"], host=host,
strict_slashes=strict_slashes, version=version)
strict_slashes=strict_slashes, version=version,
name=name)
def post(self, uri, host=None, strict_slashes=False, stream=False,
version=None):
def post(self, uri, host=None, strict_slashes=None, stream=False,
version=None, name=None):
return self.route(uri, methods=["POST"], host=host,
strict_slashes=strict_slashes, stream=stream,
version=version)
version=version, name=name)
def put(self, uri, host=None, strict_slashes=False, stream=False,
version=None):
def put(self, uri, host=None, strict_slashes=None, stream=False,
version=None, name=None):
return self.route(uri, methods=["PUT"], host=host,
strict_slashes=strict_slashes, stream=stream,
version=version)
version=version, name=name)
def head(self, uri, host=None, strict_slashes=False, version=None):
def head(self, uri, host=None, strict_slashes=None, version=None,
name=None):
return self.route(uri, methods=["HEAD"], host=host,
strict_slashes=strict_slashes, version=version)
strict_slashes=strict_slashes, version=version,
name=name)
def options(self, uri, host=None, strict_slashes=False, version=None):
def options(self, uri, host=None, strict_slashes=None, version=None,
name=None):
return self.route(uri, methods=["OPTIONS"], host=host,
strict_slashes=strict_slashes, version=version)
strict_slashes=strict_slashes, version=version,
name=name)
def patch(self, uri, host=None, strict_slashes=False, stream=False,
version=None):
def patch(self, uri, host=None, strict_slashes=None, stream=False,
version=None, name=None):
return self.route(uri, methods=["PATCH"], host=host,
strict_slashes=strict_slashes, stream=stream,
version=version)
version=version, name=name)
def delete(self, uri, host=None, strict_slashes=False, version=None):
def delete(self, uri, host=None, strict_slashes=None, version=None,
name=None):
return self.route(uri, methods=["DELETE"], host=host,
strict_slashes=strict_slashes, version=version)
strict_slashes=strict_slashes, version=version,
name=name)

11
sanic/config.py
View File

@ -131,7 +131,8 @@ class Config(dict):
self.GRACEFUL_SHUTDOWN_TIMEOUT = 15.0 # 15 sec
if load_env:
self.load_environment_vars()
prefix = SANIC_PREFIX if load_env is True else load_env
self.load_environment_vars(prefix=prefix)
def __getattr__(self, attr):
try:
@ -195,14 +196,14 @@ class Config(dict):
if key.isupper():
self[key] = getattr(obj, key)
def load_environment_vars(self):
def load_environment_vars(self, prefix=SANIC_PREFIX):
"""
Looks for any ``SANIC_`` prefixed environment variables and applies
Looks for prefixed environment variables and applies
them to the configuration if present.
"""
for k, v in os.environ.items():
if k.startswith(SANIC_PREFIX):
_, config_key = k.split(SANIC_PREFIX, 1)
if k.startswith(prefix):
_, config_key = k.split(prefix, 1)
try:
self[config_key] = int(v)
except ValueError:

29
sanic/exceptions.py
View File

@ -209,6 +209,7 @@ class Unauthorized(SanicException):
Unauthorized exception (401 HTTP status code).
:param message: Message describing the exception.
:param status_code: HTTP Status code.
:param scheme: Name of the authentication scheme to be used.
When present, kwargs is used to complete the WWW-Authentication header.
@ -216,11 +217,13 @@ class Unauthorized(SanicException):
Examples::
# With a Basic auth-scheme, realm MUST be present:
raise Unauthorized("Auth required.", "Basic", realm="Restricted Area")
raise Unauthorized("Auth required.",
scheme="Basic",
realm="Restricted Area")
# With a Digest auth-scheme, things are a bit more complicated:
raise Unauthorized("Auth required.",
"Digest",
scheme="Digest",
realm="Restricted Area",
qop="auth, auth-int",
algorithm="MD5",
@ -228,20 +231,24 @@ class Unauthorized(SanicException):
opaque="zyxwvu")
# With a Bearer auth-scheme, realm is optional so you can write:
raise Unauthorized("Auth required.", "Bearer")
raise Unauthorized("Auth required.", scheme="Bearer")
# or, if you want to specify the realm:
raise Unauthorized("Auth required.", "Bearer", realm="Restricted Area")
raise Unauthorized("Auth required.",
scheme="Bearer",
realm="Restricted Area")
"""
def __init__(self, message, scheme, **kwargs):
super().__init__(message)
def __init__(self, message, status_code=None, scheme=None, **kwargs):
super().__init__(message, status_code)
values = ["{!s}={!r}".format(k, v) for k, v in kwargs.items()]
challenge = ', '.join(values)
# if auth-scheme is specified, set "WWW-Authenticate" header
if scheme is not None:
values = ["{!s}={!r}".format(k, v) for k, v in kwargs.items()]
challenge = ', '.join(values)
self.headers = {
"WWW-Authenticate": "{} {}".format(scheme, challenge).rstrip()
}
self.headers = {
"WWW-Authenticate": "{} {}".format(scheme, challenge).rstrip()
}
def abort(status_code, message=None):

53
sanic/router.py
View File

@ -67,6 +67,7 @@ class Router:
def __init__(self):
self.routes_all = {}
self.routes_names = {}
self.routes_static = {}
self.routes_dynamic = defaultdict(list)
self.routes_always_check = []
@ -99,7 +100,7 @@ class Router:
return name, _type, pattern
def add(self, uri, methods, handler, host=None, strict_slashes=False,
version=None):
version=None, name=None):
"""Add a handler to the route list
:param uri: path to match
@ -118,29 +119,28 @@ class Router:
else:
uri = "/".join(["/v{}".format(str(version)), uri])
# add regular version
self._add(uri, methods, handler, host)
self._add(uri, methods, handler, host, name)
if strict_slashes:
return
# Add versions with and without trailing /
slash_is_missing = (
not uri[-1] == '/'
and not self.routes_all.get(uri + '/', False)
not uri[-1] == '/' and not self.routes_all.get(uri + '/', False)
)
without_slash_is_missing = (
uri[-1] == '/'
and not self.routes_all.get(uri[:-1], False)
and not uri == '/'
uri[-1] == '/' and not
self.routes_all.get(uri[:-1], False) and not
uri == '/'
)
# add version with trailing slash
if slash_is_missing:
self._add(uri + '/', methods, handler, host)
self._add(uri + '/', methods, handler, host, name)
# add version without trailing slash
elif without_slash_is_missing:
self._add(uri[:-1], methods, handler, host)
self._add(uri[:-1], methods, handler, host, name)
def _add(self, uri, methods, handler, host=None):
def _add(self, uri, methods, handler, host=None, name=None):
"""Add a handler to the route list
:param uri: path to match
@ -161,7 +161,7 @@ class Router:
"host strings, not {!r}".format(host))
for host_ in host:
self.add(uri, methods, handler, host_)
self.add(uri, methods, handler, host_, name)
return
# Dict for faster lookups of if method allowed
@ -229,22 +229,26 @@ class Router:
else:
route = self.routes_all.get(uri)
# prefix the handler name with the blueprint name
# if available
if hasattr(handler, '__blueprintname__'):
handler_name = '{}.{}'.format(
handler.__blueprintname__, name or handler.__name__)
else:
handler_name = name or getattr(handler, '__name__', None)
if route:
route = merge_route(route, methods, handler)
else:
# prefix the handler name with the blueprint name
# if available
if hasattr(handler, '__blueprintname__'):
handler_name = '{}.{}'.format(
handler.__blueprintname__, handler.__name__)
else:
handler_name = getattr(handler, '__name__', None)
route = Route(
handler=handler, methods=methods, pattern=pattern,
parameters=parameters, name=handler_name, uri=uri)
self.routes_all[uri] = route
pairs = self.routes_names.get(handler_name)
if not (pairs and (pairs[0] + '/' == uri or uri + '/' == pairs[0])):
self.routes_names[handler_name] = (uri, route)
if properties['unhashable']:
self.routes_always_check.append(route)
elif parameters:
@ -265,6 +269,11 @@ class Router:
uri = host + uri
try:
route = self.routes_all.pop(uri)
for handler_name, pairs in self.routes_names.items():
if pairs[0] == uri:
self.routes_names.pop(handler_name)
break
except KeyError:
raise RouteDoesNotExist("Route was not registered: {}".format(uri))
@ -289,11 +298,7 @@ class Router:
if not view_name:
return (None, None)
for uri, route in self.routes_all.items():
if route.name == view_name:
return uri, route
return (None, None)
return self.routes_names.get(view_name, (None, None))
def get(self, request):
"""Get a request handler based on the URL of the request, or raises an

59
tests/test_blueprints.py
View File

@ -78,6 +78,65 @@ def test_bp_strict_slash():
request, response = app.test_client.post('/post')
assert response.status == 404
def test_bp_strict_slash_default_value():
app = Sanic('test_route_strict_slash')
bp = Blueprint('test_text', strict_slashes=True)
@bp.get('/get')
def handler(request):
return text('OK')
@bp.post('/post/')
def handler(request):
return text('OK')
app.blueprint(bp)
request, response = app.test_client.get('/get/')
assert response.status == 404
request, response = app.test_client.post('/post')
assert response.status == 404
def test_bp_strict_slash_without_passing_default_value():
app = Sanic('test_route_strict_slash')
bp = Blueprint('test_text')
@bp.get('/get')
def handler(request):
return text('OK')
@bp.post('/post/')
def handler(request):
return text('OK')
app.blueprint(bp)
request, response = app.test_client.get('/get/')
assert response.text == 'OK'
request, response = app.test_client.post('/post')
assert response.text == 'OK'
def test_bp_strict_slash_default_value_can_be_overwritten():
app = Sanic('test_route_strict_slash')
bp = Blueprint('test_text', strict_slashes=True)
@bp.get('/get', strict_slashes=False)
def handler(request):
return text('OK')
@bp.post('/post/', strict_slashes=False)
def handler(request):
return text('OK')
app.blueprint(bp)
request, response = app.test_client.get('/get/')
assert response.text == 'OK'
request, response = app.test_client.post('/post')
assert response.text == 'OK'
def test_bp_with_url_prefix():
app = Sanic('test_text')

10
tests/test_config.py
View File

@ -19,15 +19,21 @@ def test_load_from_object():
def test_auto_load_env():
environ["SANIC_TEST_ANSWER"] = "42"
app = Sanic()
assert app.config.TEST_ANSWER == "42"
assert app.config.TEST_ANSWER == 42
del environ["SANIC_TEST_ANSWER"]
def test_auto_load_env():
def test_dont_load_env():
environ["SANIC_TEST_ANSWER"] = "42"
app = Sanic(load_env=False)
assert getattr(app.config, 'TEST_ANSWER', None) == None
del environ["SANIC_TEST_ANSWER"]
def test_load_env_prefix():
environ["MYAPP_TEST_ANSWER"] = "42"
app = Sanic(load_env='MYAPP_')
assert app.config.TEST_ANSWER == 42
del environ["MYAPP_TEST_ANSWER"]
def test_load_from_file():
app = Sanic('test_load_from_file')
config = b"""

20
tests/test_exceptions.py
View File

@ -31,14 +31,18 @@ def exception_app():
def handler_403(request):
raise Forbidden("Forbidden")
@app.route('/401')
def handler_401(request):
raise Unauthorized("Unauthorized")
@app.route('/401/basic')
def handler_401_basic(request):
raise Unauthorized("Unauthorized", "Basic", realm="Sanic")
raise Unauthorized("Unauthorized", scheme="Basic", realm="Sanic")
@app.route('/401/digest')
def handler_401_digest(request):
raise Unauthorized("Unauthorized",
"Digest",
scheme="Digest",
realm="Sanic",
qop="auth, auth-int",
algorithm="MD5",
@ -47,12 +51,16 @@ def exception_app():
@app.route('/401/bearer')
def handler_401_bearer(request):
raise Unauthorized("Unauthorized", "Bearer")
raise Unauthorized("Unauthorized", scheme="Bearer")
@app.route('/invalid')
def handler_invalid(request):
raise InvalidUsage("OK")
@app.route('/abort/401')
def handler_invalid(request):
abort(401)
@app.route('/abort')
def handler_invalid(request):
abort(500)
@ -124,6 +132,9 @@ def test_forbidden_exception(exception_app):
def test_unauthorized_exception(exception_app):
"""Test the built-in Unauthorized exception"""
request, response = exception_app.test_client.get('/401')
assert response.status == 401
request, response = exception_app.test_client.get('/401/basic')
assert response.status == 401
assert response.headers.get('WWW-Authenticate') is not None
@ -186,5 +197,8 @@ def test_exception_in_exception_handler_debug_off(exception_app):
def test_abort(exception_app):
"""Test the abort function"""
request, response = exception_app.test_client.get('/abort/401')
assert response.status == 401
request, response = exception_app.test_client.get('/abort')
assert response.status == 500

388
tests/test_named_routes.py Normal file
View File

@ -0,0 +1,388 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import asyncio
import pytest
from sanic import Sanic
from sanic.blueprints import Blueprint
from sanic.response import text
from sanic.exceptions import URLBuildError
from sanic.constants import HTTP_METHODS
# ------------------------------------------------------------ #
# UTF-8
# ------------------------------------------------------------ #
@pytest.mark.parametrize('method', HTTP_METHODS)
def test_versioned_named_routes_get(method):
app = Sanic('test_shorhand_routes_get')
bp = Blueprint('test_bp', url_prefix='/bp')
method = method.lower()
route_name = 'route_{}'.format(method)
route_name2 = 'route2_{}'.format(method)
func = getattr(app, method)
if callable(func):
@func('/{}'.format(method), version=1, name=route_name)
def handler(request):
return text('OK')
else:
print(func)
raise
func = getattr(bp, method)
if callable(func):
@func('/{}'.format(method), version=1, name=route_name2)
def handler2(request):
return text('OK')
else:
print(func)
raise
app.blueprint(bp)
assert app.router.routes_all['/v1/{}'.format(method)].name == route_name
route = app.router.routes_all['/v1/bp/{}'.format(method)]
assert route.name == 'test_bp.{}'.format(route_name2)
assert app.url_for(route_name) == '/v1/{}'.format(method)
url = app.url_for('test_bp.{}'.format(route_name2))
assert url == '/v1/bp/{}'.format(method)
with pytest.raises(URLBuildError):
app.url_for('handler')
def test_shorthand_default_routes_get():
app = Sanic('test_shorhand_routes_get')
@app.get('/get')
def handler(request):
return text('OK')
assert app.router.routes_all['/get'].name == 'handler'
assert app.url_for('handler') == '/get'
def test_shorthand_named_routes_get():
app = Sanic('test_shorhand_routes_get')
bp = Blueprint('test_bp', url_prefix='/bp')
@app.get('/get', name='route_get')
def handler(request):
return text('OK')
@bp.get('/get', name='route_bp')
def handler2(request):
return text('Blueprint')
app.blueprint(bp)
assert app.router.routes_all['/get'].name == 'route_get'
assert app.url_for('route_get') == '/get'
with pytest.raises(URLBuildError):
app.url_for('handler')
assert app.router.routes_all['/bp/get'].name == 'test_bp.route_bp'
assert app.url_for('test_bp.route_bp') == '/bp/get'
with pytest.raises(URLBuildError):
app.url_for('test_bp.handler2')
def test_shorthand_named_routes_post():
app = Sanic('test_shorhand_routes_post')
@app.post('/post', name='route_name')
def handler(request):
return text('OK')
assert app.router.routes_all['/post'].name == 'route_name'
assert app.url_for('route_name') == '/post'
with pytest.raises(URLBuildError):
app.url_for('handler')
def test_shorthand_named_routes_put():
app = Sanic('test_shorhand_routes_put')
@app.put('/put', name='route_put')
def handler(request):
assert request.stream is None
return text('OK')
assert app.is_request_stream is False
assert app.router.routes_all['/put'].name == 'route_put'
assert app.url_for('route_put') == '/put'
with pytest.raises(URLBuildError):
app.url_for('handler')
def test_shorthand_named_routes_delete():
app = Sanic('test_shorhand_routes_delete')
@app.delete('/delete', name='route_delete')
def handler(request):
assert request.stream is None
return text('OK')
assert app.is_request_stream is False
assert app.router.routes_all['/delete'].name == 'route_delete'
assert app.url_for('route_delete') == '/delete'
with pytest.raises(URLBuildError):
app.url_for('handler')
def test_shorthand_named_routes_patch():
app = Sanic('test_shorhand_routes_patch')
@app.patch('/patch', name='route_patch')
def handler(request):
assert request.stream is None
return text('OK')
assert app.is_request_stream is False
assert app.router.routes_all['/patch'].name == 'route_patch'
assert app.url_for('route_patch') == '/patch'
with pytest.raises(URLBuildError):
app.url_for('handler')
def test_shorthand_named_routes_head():
app = Sanic('test_shorhand_routes_head')
@app.head('/head', name='route_head')
def handler(request):
assert request.stream is None
return text('OK')
assert app.is_request_stream is False
assert app.router.routes_all['/head'].name == 'route_head'
assert app.url_for('route_head') == '/head'
with pytest.raises(URLBuildError):
app.url_for('handler')
def test_shorthand_named_routes_options():
app = Sanic('test_shorhand_routes_options')
@app.options('/options', name='route_options')
def handler(request):
assert request.stream is None
return text('OK')
assert app.is_request_stream is False
assert app.router.routes_all['/options'].name == 'route_options'
assert app.url_for('route_options') == '/options'
with pytest.raises(URLBuildError):
app.url_for('handler')
def test_named_static_routes():
app = Sanic('test_dynamic_route')
@app.route('/test', name='route_test')
async def handler1(request):
return text('OK1')
@app.route('/pizazz', name='route_pizazz')
async def handler2(request):
return text('OK2')
assert app.router.routes_all['/test'].name == 'route_test'
assert app.router.routes_static['/test'].name == 'route_test'
assert app.url_for('route_test') == '/test'
with pytest.raises(URLBuildError):
app.url_for('handler1')
assert app.router.routes_all['/pizazz'].name == 'route_pizazz'
assert app.router.routes_static['/pizazz'].name == 'route_pizazz'
assert app.url_for('route_pizazz') == '/pizazz'
with pytest.raises(URLBuildError):
app.url_for('handler2')
def test_named_dynamic_route():
app = Sanic('test_dynamic_route')
results = []
@app.route('/folder/<name>', name='route_dynamic')
async def handler(request, name):
results.append(name)
return text('OK')
assert app.router.routes_all['/folder/<name>'].name == 'route_dynamic'
assert app.url_for('route_dynamic', name='test') == '/folder/test'
with pytest.raises(URLBuildError):
app.url_for('handler')
def test_dynamic_named_route_regex():
app = Sanic('test_dynamic_route_regex')
@app.route('/folder/<folder_id:[A-Za-z0-9]{0,4}>', name='route_re')
async def handler(request, folder_id):
return text('OK')
route = app.router.routes_all['/folder/<folder_id:[A-Za-z0-9]{0,4}>']
assert route.name == 'route_re'
assert app.url_for('route_re', folder_id='test') == '/folder/test'
with pytest.raises(URLBuildError):
app.url_for('handler')
def test_dynamic_named_route_path():
app = Sanic('test_dynamic_route_path')
@app.route('/<path:path>/info', name='route_dynamic_path')
async def handler(request, path):
return text('OK')
route = app.router.routes_all['/<path:path>/info']
assert route.name == 'route_dynamic_path'
assert app.url_for('route_dynamic_path', path='path/1') == '/path/1/info'
with pytest.raises(URLBuildError):
app.url_for('handler')
def test_dynamic_named_route_unhashable():
app = Sanic('test_dynamic_route_unhashable')
@app.route('/folder/<unhashable:[A-Za-z0-9/]+>/end/',
name='route_unhashable')
async def handler(request, unhashable):
return text('OK')
route = app.router.routes_all['/folder/<unhashable:[A-Za-z0-9/]+>/end/']
assert route.name == 'route_unhashable'
url = app.url_for('route_unhashable', unhashable='test/asdf')
assert url == '/folder/test/asdf/end'
with pytest.raises(URLBuildError):
app.url_for('handler')
def test_websocket_named_route():
app = Sanic('test_websocket_route')
ev = asyncio.Event()
@app.websocket('/ws', name='route_ws')
async def handler(request, ws):
assert ws.subprotocol is None
ev.set()
assert app.router.routes_all['/ws'].name == 'route_ws'
assert app.url_for('route_ws') == '/ws'
with pytest.raises(URLBuildError):
app.url_for('handler')
def test_websocket_named_route_with_subprotocols():
app = Sanic('test_websocket_route')
results = []
@app.websocket('/ws', subprotocols=['foo', 'bar'], name='route_ws')
async def handler(request, ws):
results.append(ws.subprotocol)
assert app.router.routes_all['/ws'].name == 'route_ws'
assert app.url_for('route_ws') == '/ws'
with pytest.raises(URLBuildError):
app.url_for('handler')
def test_static_add_named_route():
app = Sanic('test_static_add_route')
async def handler1(request):
return text('OK1')
async def handler2(request):
return text('OK2')
app.add_route(handler1, '/test', name='route_test')
app.add_route(handler2, '/test2', name='route_test2')
assert app.router.routes_all['/test'].name == 'route_test'
assert app.router.routes_static['/test'].name == 'route_test'
assert app.url_for('route_test') == '/test'
with pytest.raises(URLBuildError):
app.url_for('handler1')
assert app.router.routes_all['/test2'].name == 'route_test2'
assert app.router.routes_static['/test2'].name == 'route_test2'
assert app.url_for('route_test2') == '/test2'
with pytest.raises(URLBuildError):
app.url_for('handler2')
def test_dynamic_add_named_route():
app = Sanic('test_dynamic_add_route')
results = []
async def handler(request, name):
results.append(name)
return text('OK')
app.add_route(handler, '/folder/<name>', name='route_dynamic')
assert app.router.routes_all['/folder/<name>'].name == 'route_dynamic'
assert app.url_for('route_dynamic', name='test') == '/folder/test'
with pytest.raises(URLBuildError):
app.url_for('handler')
def test_dynamic_add_named_route_unhashable():
app = Sanic('test_dynamic_add_route_unhashable')
async def handler(request, unhashable):
return text('OK')
app.add_route(handler, '/folder/<unhashable:[A-Za-z0-9/]+>/end/',
name='route_unhashable')
route = app.router.routes_all['/folder/<unhashable:[A-Za-z0-9/]+>/end/']
assert route.name == 'route_unhashable'
url = app.url_for('route_unhashable', unhashable='folder1')
assert url == '/folder/folder1/end'
with pytest.raises(URLBuildError):
app.url_for('handler')
def test_overload_routes():
app = Sanic('test_dynamic_route')
@app.route('/overload', methods=['GET'], name='route_first')
async def handler1(request):
return text('OK1')
@app.route('/overload', methods=['POST', 'PUT'], name='route_second')
async def handler1(request):
return text('OK2')
request, response = app.test_client.get(app.url_for('route_first'))
assert response.text == 'OK1'
request, response = app.test_client.post(app.url_for('route_first'))
assert response.text == 'OK2'
request, response = app.test_client.put(app.url_for('route_first'))
assert response.text == 'OK2'
request, response = app.test_client.get(app.url_for('route_second'))
assert response.text == 'OK1'
request, response = app.test_client.post(app.url_for('route_second'))
assert response.text == 'OK2'
request, response = app.test_client.put(app.url_for('route_second'))
assert response.text == 'OK2'
assert app.router.routes_all['/overload'].name == 'route_first'
with pytest.raises(URLBuildError):
app.url_for('handler1')
assert app.url_for('route_first') == '/overload'
assert app.url_for('route_second') == app.url_for('route_first')

30
tests/test_routes.py
View File

@ -71,6 +71,36 @@ def test_route_strict_slash():
request, response = app.test_client.post('/post')
assert response.status == 404
def test_route_strict_slash_default_value():
app = Sanic('test_route_strict_slash', strict_slashes=True)
@app.get('/get')
def handler(request):
return text('OK')
request, response = app.test_client.get('/get/')
assert response.status == 404
def test_route_strict_slash_without_passing_default_value():
app = Sanic('test_route_strict_slash')
@app.get('/get')
def handler(request):
return text('OK')
request, response = app.test_client.get('/get/')
assert response.text == 'OK'
def test_route_strict_slash_default_value_can_be_overwritten():
app = Sanic('test_route_strict_slash', strict_slashes=True)
@app.get('/get', strict_slashes=False)
def handler(request):
return text('OK')
request, response = app.test_client.get('/get/')
assert response.text == 'OK'
def test_route_optional_slash():
app = Sanic('test_route_optional_slash')