python框架sanic_Sanic框架
1. 入门
Sanic 是一款类似Flask的Web服务器，它运行在Python 3.5+上。
除了与Flask功能类似之外，它还支持异步请求处理，这意味着你可以使用Python3.5 中新的异步/等待语法，使你的程序运行更加快速。
1.1 简单起步
from sanic import Sanic
from sanic.response import json
app = Sanic()
@app.route("/")
async def test(request):
return json({"hello": "world"})
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8000)
保存到main.py文件，运行文件python3 main.py；
打开URLhttp://0.0.0.0:8000，可以看到网页显示 Hello World信息。
2. 路由
路由允许用户为不同的URL地址指定处理的函数。
一个基本的路由就像下面的例子，而app就是Sanic类的一个实例。
from sanic.response import json
@app.route("/")
async def test(request):
return json({ "hello": "world" })
当地址http://server.url/被访问时(服务的基础地址)，根地址/就会被路由匹配一个定义了返回JSON对象的test函数。
必须使用async def语法定义函数，来保证其可以进行异步处理。
2.1 请求参数
Sanic的基础路由支持请求参数的操作。
如果需要指定参数，请使用尖括号将指定参数括起来。请求参数将作为路由函数的关键字参数。
from sanic.response import text
@app.route('/tag/')
async def tag_handler(request, tag):
return text('Tag - {}'.format(tag))
如果需要指定添加的参数的类型，则要在参数名字后面添加:type指定参数类型。如果参数与指定的参数类型不匹配，则Sanic会抛出NotFound的异常，从而导致页面出现404: Page not found的错误。
from sanic.response import text
@app.route('/number/')
async def integer_handler(request, integer_arg):
return text('Integer - {}'.format(integer_arg))
@app.route('/number/')
async def number_handler(request, number_arg):
return text('Number - {}'.format(number_arg))
@app.route('/person/')
async def person_handler(request, name):
return text('Person - {}'.format(name))
@app.route('/folder/')
async def folder_handler(request, folder_id):
return text('Folder - {}'.format(folder_id))
2.2 HTTP 请求类型
默认情况下，一个路由会定义一个仅仅适用于URL的GET请求。然而，@app.route装饰器接受一个可选的参数methods，它允许定义的函数使用列表中任何一个的HTTP方法。
from sanic.response import text
@app.route('/post', methods=['POST'])
async def post_handler(request):
return text('POST request - {}'.format(request.json))
@app.route('/get', methods=['GET'])
async def get_handler(request):
return text('GET request - {}'.format(request.args))
这里还有一个可选的host参数(列表或是字符串)。它限制了给主机的路由。如果存在一个没有主机的路由，它将是一个默认值。
@app.route('/get', methods=['GET'], host='example.com')
async def get_handler(request):
return text('GET request - {}'.format(request.args))
# if the host header doesn't match example.com, this route will be used
@app.route('/get', methods=['GET'])
async def get_handler(request):
return text('GET request in default - {}'.format(request.args))
这里还有一种快速使用装饰器的方法：
from sanic.response import text
@app.post('/post')
async def post_handler(request):
return text('POST request - {}'.format(request.json))
@app.get('/get')
async def get_handler(request):
return text('GET request - {}'.format(request.args))
2.3 add_route方法
就像上文提到的，路由通常使用@app.route装饰器进行添加的。但是，这个装饰器只是app.add_route方法的一个封装。它看起来像下面这样：
from sanic.response import text
# Define the handler functions
async def handler1(request):
return text('OK')
async def handler2(request, name):
return text('Folder - {}'.format(name))
async def person_handler2(request, name):
return text('Person - {}'.format(name))
# Add each handler function as a route
app.add_route(handler1, '/test')
app.add_route(handler2, '/folder/')
app.add_route(person_handler2, '/person/', methods=['GET'])
2.4 利用url_for生成URL
Sanic提供了一个根据处理函数名字生成URL的方法url_for。在应用中，它使用处理程序的名字来有效避免使用实际的网络路径。
@app.route('/')
async def index(request):
# generate a URL for the endpoint `post_handler`
url = app.url_for('post_handler', post_id=5)
# the URL is `/posts/5`, redirect to it
return redirect(url)
@app.route('/posts/')
async def post_handler(request, post_id):
return text('Post - {}'.format(post_id))
使用url_for需要注意的是:
传递给url_for的关键字如果不是请求参数，将包含在URL 的查询字符串中。
url = app.url_for('post_handler', post_id=5, arg_one='one', arg_two='two')
# /posts/5?arg_one=one&arg_two=two
可以传递多个参数给url_for函数。
url = app.url_for('post_handler', post_id=5, arg_one=['one', 'two'])
# /posts/5?arg_one=one&arg_one=two
还可以传递一些特殊的参数给url_for方法来构造一些特殊的URL，诸如：_anchor,_external,_scheme,_server。
url = app.url_for('post_handler', post_id=5, arg_one='one', _anchor='anchor')
# /posts/5?arg_one=one#anchor
url = app.url_for('post_handler', post_id=5, arg_one='one', _external=True)
# //server/posts/5?arg_one=one
# _external requires passed argument _server or SERVER_NAME in app.config or url will be same as no _external
url = app.url_for('post_handler', post_id=5, arg_one='one', _scheme='http', _external=True)
# http://server/posts/5?arg_one=one
# when specifying _scheme, _external must be True
# you can pass all special arguments one time
url = app.url_for('post_handler', post_id=5, arg_one=['one', 'two'], arg_two=2, _anchor='anchor', _scheme='http', _external=True, _server='another_server:8888')
# http://another_server:8888/posts/5?arg_one=one&arg_one=two&arg_two=2#anchor
所有的参数都必须正确地传递给url_for方法来构造URL。如果未提供参数或指定参数不匹配，将抛出URLBuildError的错误。
2.5 WebSocket 路由
使用@app.websocket装饰器定义WebSocket协议的路由。
@app.websocket('/feed')
async def feed(request, ws):
while True:
data = 'hello!'
print('Sending: ' + data)
await ws.send(data)
data = await ws.recv()
print('Received: ' + data)
或者，使用app.add_websocket_route方法来代替@app.websocket装饰器。
async def feed(request, ws):
pass
app.add_websocket_route(my_websocket_handler, '/feed')
WebSocket路由的处理程序将请求作为第一个参数传递，并将WebSocket协议对象作为第二个参数传递。而协议对象具有send和recv两个方法来进行数据的传送和接收。
3. 请求数据
当接收端接收到一个HTTP请求的时候，路由函数就会传递一个Request对象。
以下的变量可以作为Request对象的属性进行访问。
json(任何类型)-JSON格式的数据
from sanic.response import json
@app.route("/json")
def post_json(request):
return json({ "received": True, "message": request.json })
arg(dict类型)-查询字符串变量。一个查询的字符串是部分的URL，类似于?key1=value1&key2=value2，如果要解析这个URL，那么arg字典看起来就像{'key1': ['value1'], 'key2': ['value2']}，这个请求将使用query_string变量来保存未解析的字符串的值。
from sanic.response import json
@app.route("/query_string")
def query_string(request):
return json({ "parsed": True, "args": request.args, "url": request.url, "query_string": request.query_string })
raw_args(dict类型)-在许多情况下，你只需要访问一段很小的url参数。对于前文的URL的?key1=value1&key2=value2，raw_args字典看起来就像这样{'key1': 'value1', 'key2': 'value2'}。
files(File对象的字典)-具有名称，正文和类型的文件列表。
from sanic.response import json
@app.route("/files")
def post_json(request):
test_file = request.files.get('test')
file_parameters = {
'body': test_file.body,
'name': test_file.name,
'type': test_file.type,
}
return json({ "received": True, "file_names": request.files.keys(), "test_file_parameters": file_parameters })
form(dict类型)-发布表单数据
from sanic.response import json
@app.route("/form")
def post_json(request):
return json({ "received": True, "form_data": request.form, "test": request.form.get('test') })
body(字节类型)-发布正文。这个属性允许检索请求的原始数据，而无需理会数据的类型。
from sanic.response import text
@app.route("/users", methods=["POST",])
def create_user(request):
return text("You are trying to create a user with the following POST: %s" % request.body)
ip(str类型)-请求者的IP地址
app-对正在处理此请求的Sanic应用程序对象的引用。在无法访问全局app对象或其它处理程序的时候很有用。
from sanic.response import json
from sanic import Blueprint
bp = Blueprint('my_blueprint')
@bp.route('/')
async def bp_root(request):
if request.app.config['DEBUG']:
return json({'status': 'debug'})
else:
return json({'status': 'production'})
url:完整的请求URL，即：http://localhost:8000/posts/1/?foo=bar。
scheme:与请求相关联的URL方案，http或是https。
host:与请求相关联的主机，localhost:8080。
path:请求的地址/post/1
query_string:请求的查询字符串，foo=bar或是空字符串''。
3.1 使用get和getlist来访问值
请求的属性实际上是返回一个dict的子类RequestParameters。在使用这个对象的主要区别是get和getlist方法的不同。
get(key, default=None) 当给定key的值是一个列表的时候，只返回第一个项目。
getlist(key, default=None) 返回整个列表。
from sanic.request import RequestParameters
args = RequestParameters()
args['titles'] = ['Post 1', 'Post 2']
args.get('titles') # => 'Post 1'
args.getlist('titles') # => ['Post 1', 'Post 2']
4. 响应
使用sanic.response模块中的函数来创建响应。
4.1 纯文本
from sanic import response
@app.route('/text')
def handle_request(request):
return response.text('Hello world!')
4.2 HTML
from sanic import response
@app.route('/html')
def handle_request(request):
return response.html('
Hello world!
')4.3 JSON
from sanic import response
@app.route('/json')
def handle_request(request):
return response.json({'message': 'Hello world!'})
4.4 文件
from sanic import response
@app.route('/file')
async def handle_request(request):
return await response.file('/srv/www/whatever.png')
4.5 Streaming
from sanic import response
@app.route("/streaming")
async def index(request):
async def streaming_fn(response):
response.write('foo')
response.write('bar')
return response.stream(streaming_fn, content_type='text/plain')
4.6 重定向
from sanic import response
@app.route('/redirect')
def handle_request(request):
return response.redirect('/json')
4.7 元数据
响应未编码的文本
from sanic import response
@app.route('/raw')
def handle_request(request):
return response.raw('raw data')
4.8 修改标题或状态
要修改标题或状态，请将标题或状态参数传递给这些函数：
from sanic import response
@app.route('/json')
def handle_request(request):
return response.json(
{'message': 'Hello world!'},
headers={'X-Served-By': 'sanic'},
status=200
)
5. 静态文件
静态文件和目录，例如图像文件是在Sanic创建app.static时候提供的。这个方法采用一个URL地址和一个文件名。然后通过给定的端点访问指定的文件。
from sanic import Sanic
app = Sanic(__name__)
# Serves files from the static folder to the URL /static
app.static('/static', './static')
# Serves the file /home/ubuntu/test.png when the URL /the_best.png
# is requested
app.static('/the_best.png', '/home/ubuntu/test.png')
app.run(host="0.0.0.0", port=8000)
目前还无法利用url_for为静态文件创建一个URL地址。
6. 异常
异常可以从请求处理程序中抛出，并由Sanic自动处理。异常将消息作为第一个参数，也可以在HTTP响应中传回状态代码。
6.1 抛出异常
要抛出一个异常，只需要从sanic.exceptions中导入与raise相关的异常。
from sanic.exceptions import ServerError
@app.route('/killme')
def i_am_ready_to_die(request):
raise ServerError("Something bad happened", status_code=500)
6.2 处理异常
如果需要覆盖Sanic对异常的默认处理，就需要使用@app.exception装饰器。装饰器期望使用一个异常列表来处理参数。你可以传递一个SanicException来捕捉它们。装饰器异常处理函数必须使用Request和Exception对象来作为参数。
from sanic.response import text
from sanic.exceptions import NotFound
@app.exception(NotFound)
def ignore_404s(request, exception):
return text("Yep, I totally found the page: {}".format(request.url))
6.3 有用的异常
一些有用的异常如下：
NotFound: 找不到合适的路由请求。
ServerError: 服务器内部出现问题时调用。通常发生在用户代码出现错误的情况。
7. 中间件和监听器
中间件是在向服务器请求之前或之后执行的功能。它们可用来修改用户自定义处理函数的请求或响应。
另外，Sanic提供程序监听器来运行应用程序生命周期中各个不同点的代码。
7.1 中间件
这里有两种不同类型的中间件：请求request和响应response。 都是使用@app.middleware装饰器进行声明的，利用'request'或'response'字符串来表示其参数类型。
最简单的中间件不修改任何的请求或响应：
@app.middleware('request')
async def print_on_request(request):
print("I print when a request is received by the server")
@app.middleware('response')
async def print_on_response(request, response):
print("I print when a response is returned by the server")
7.2 修改请求或响应
只要请求或修改不返回任何值，中间件就可以修改给定的请求或响应。下面的示例即是一个简单的示范：
app = Sanic(__name__)
@app.middleware('response')
async def custom_banner(request, response):
response.headers["Server"] = "Fake-Server"
@app.middleware('response')
async def prevent_xss(request, response):
response.headers["x-xss-protection"] = "1; mode=block"
app.run(host="0.0.0.0", port=8000)
上面的代码将按顺序应用两个中间件。首先，中间件custom_banner将HTTP响应头服务器更改为Fake-Server，第二个中间件prevent_xss将添加HTTP头以防止跨站点脚本(XSS)攻击。这两个函数在用户函数返回响应之后调用。
如果中间件返回一个HTTPResponse对象，这个请求将停止处理并返回响应。如果这个请求在相关用户路由处理到达之前发生，则不会被调用该处理程序。返回的响应还会阻止进一步的中间件运行。
@app.middleware('request')
async def halt_request(request):
return text('I halted the request')
@app.middleware('response')
async def halt_response(request, response):
return text('I halted the response')
7.3 监听器
如果你想要在服务启动或关闭时执行启动/拆卸代码，可以使用以下的监听器：
before_server_start
after_server_start
before_server_stop
after_server_stop
这些监听器在接收app对象和asyncio循环的函数上实现为装饰器。如下所示：
@app.listener('before_server_start')
async def setup_db(app, loop):
app.db = await db_setup()
@app.listener('after_server_start')
async def notify_server_started(app, loop):
print('Server successfully started!')
@app.listener('before_server_stop')
async def notify_server_stopping(app, loop):
print('Server shutting down!')
@app.listener('after_server_stop')
async def close_db(app, loop):
await app.db.close()
如果你想要在循环开始后安排后台允许任务，则可以使用add_task方法轻松实现。
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())
8. 蓝图
蓝图是可以用于应用程序中的子路由对象。蓝图不是向应用程序实例添加路由，而是定义了类似添加路由的方法，然后将路由以灵活且可插拔的方式注册到应用程序中。
蓝图对于较大的应用程序十分有用，你可在逻辑上将应用程序分为几个组或责任领域。
8.1 第一个蓝图
下面显示了一个非常简单的蓝图，它在你的应用程序的根目录/下注册了一个处理函数。
假设你将其保存到了my_blueprint.py文件中，则可以将其导入到你的主应用程序中。
from sanic.response import json
from sanic import Blueprint
bp = Blueprint('my_blueprint')
@bp.route('/')
async def bp_root(request):
return json({'my': 'blueprint'})
8.2 注册蓝图
蓝图必须在应用程序中注册。
from sanic import Sanic
from my_blueprint import bp
app = Sanic(__name__)
app.blueprint(bp)
app.run(host='0.0.0.0', port=8000, debug=True)
蓝图将添加到应用程序，并注册由该蓝图定义的任何路由。在此示例中，app.router中的注册路由将如下所示：
[Route(handler=, methods=None, pattern=re.compile('^/$'), parameters=[])]
8.3 使用蓝图
蓝图与应用程序有着大致相同的功能。
8.3.1 WebSocket协议路由
可以使用@bp.websocket装饰器或bp.add_websocket_route方法在蓝图上注册WebSocket处理程序。
8.3.2 中间件
使用蓝图可以在全局注册中间件。
@bp.middleware
async def print_on_request(request):
print("I am a spy")
@bp.middleware('request')
async def halt_request(request):
return text('I halted the request')
@bp.middleware('response')
async def halt_response(request, response):
return text('I halted the response')
8.3.3 异常
只利用蓝图来应用全局的异常。
@bp.exception(NotFound)
def ignore_404s(request, exception):
return text("Yep, I totally found the page: {}".format(request.url))
8.3.4 静态文件
静态文件可以在蓝图定义下提供给全局。
bp.static('/folder/to/serve', '/web/path')
8.4 启动和停止
蓝图可以在服务启动或停止之前运行功能。如果多进程模式运行(超过1个进程)，这些将在进程fork之后被触发。
可用的事件是：
before_server_start: 服务开始接受连接之前执行
after_server_start: 服务开始接受连接后执行
before_server_stop: 服务停止接受连接之前执行
after_server_stop: 服务停止并且所有请求完成后执行
bp = Blueprint('my_blueprint')
@bp.listener('before_server_start')
async def setup_connection(app, loop):
global database
database = mysql.connect(host='127.0.0.1'...)
@bp.listener('after_server_stop')
async def close_connection(app, loop):
await database.close()
8.5 用例：API版本控制
Blueprints对于API版本控制非常有用，其中一个蓝图可能指向/v1/，另一个指向/v2/。
当蓝图被初始化时，它可以使用一个可选的url_prefix参数，这个参数将被添加到蓝图上定义的所有路由上。此功能可用于实现API版本控制。
# blueprints.py
from sanic.response import text
from sanic import Blueprint
blueprint_v1 = Blueprint('v1', url_prefix='/v1')
blueprint_v2 = Blueprint('v2', url_prefix='/v2')
@blueprint_v1.route('/')
async def api_v1_root(request):
return text('Welcome to version 1 of our documentation')
@blueprint_v2.route('/')
async def api_v2_root(request):
return text('Welcome to version 2 of our documentation')
当在应用程序上注册蓝图时，路由/v1和/v2现在将指向单个蓝图，这允许为每个API版本创建子站点。
# main.py
from sanic import Sanic
from blueprints import blueprint_v1, blueprint_v2
app = Sanic(__name__)
app.blueprint(blueprint_v1, url_prefix='/v1')
app.blueprint(blueprint_v2, url_prefix='/v2')
app.run(host='0.0.0.0', port=8000, debug=True)
8.6 利用url_for生成URL
如果希望为蓝图中的路由生成URL，请记住端点名称采用的格式，。例如：
@blueprint_v1.route('/')
async def root(request):
url = app.url_for('v1.post_handler', post_id=5) # --> '/v1/post/5'
return redirect(url)
@blueprint_v1.route('/post/')
async def post_handler(request, post_id):
return text('Post {} in Blueprint V1'.format(post_id))
9. 配置
任何复杂的应用程序都需要合理的配置。不同的环境或安装的设置可能不同。
9.1 基本配置
Sanic将配置保存在config应用程序对象的属性中。配置的是一个可以使用点运算进行修改或是类似字典类型的对象。
app = Sanic('myapp')
app.config.DB_NAME = 'appdb'
app.config.DB_USER = 'appuser'
由于配置的对象实际上是一个字典，你可以使用update方法来一次性设置几个值。
db_settings = {
'DB_HOST': 'localhost',
'DB_NAME': 'appdb',
'DB_USER': 'appuser'
}
app.config.update(db_settings)
一般惯例是只有UPPERCASE配置参数。下面描述的用于仅查找类似于UPPERCASE参数加载配置的方法。
9.2 加载配置
有几种方式加载配置。
9.2.1 从环境变量加载
任何由SANIC_定义的变量都将应用于sanic配置。例如，设置SANIC_REQUEST_TIMEOUT自动加载应用程序。你可以使用load_cars将布尔值传递给Sanic构造函数来进行覆盖。
app = Sanic(load_vars=False)
9.2.2 从对象加载
如果有很多配置参数并且它们有合理的默认值，将它们放置于模块是有帮助的。
import myapp.default_settings
app = Sanic('myapp')
app.config.from_object(myapp.default_settings)
你也可以使用类或者其它的对象类型。
9.2.3 从文件加载
通常情况下，你想要从文件中加载配置参数。你可以从from_file(/path/to/config_file)来加载配置参数。然而，这需要程序知道配置文件的位置，所以你可以在环境变量中指定配置文件的路径，并让Sanic寻找配置文件并使用配置文件。
app = Sanic('myapp')
app.config.from_envvar('MYAPP_SETTINGS')
然后你可以在MYAPP_SETTINGS环境设置下运行你的应用程序：
$ MYAPP_SETTINGS=/path/to/config_file python3 myapp.py
INFO: Goin' Fast @ http://0.0.0.0:8000
配置文件是常规的Python文件，运行它们只是为了加载配置。这允许你使用任何正确的逻辑进行正确的配置。只要uppercase变量被添加到配置中，最常见的配置包括简单的键值对：
# config_file
DB_HOST = 'localhost'
DB_NAME = 'appdb'
DB_USER = 'appuser'
9.3 内置的配置参数
提供的几个预设值可以在创建应用程序的时候被覆盖：
Variable
Default
Description
REQUEST_MAX_SIZE
100000000
How big a request may be (bytes)
REQUEST_TIMEOUT
60
How long a request can take (sec)
KEEP_ALIVE
True
Disables keep-alive when False
10. Cookies
Cookies是持续保存在用户浏览器中的数据片段。Sanic可以读取和写入Cookies，并以键值对的形式保存。
10.1 读取Cookies
可以通过Request对象的cookies字典访问访问用户的cookies。
from sanic.response import text
@app.route("/cookie")
async def test(request):
test_cookie = request.cookies.get('test')
return text("Test cookie set to: {}".format(test_cookie))
10.2 写入Cookies
当返回一个响应时，可以在Response对象上设置Cookies。
from sanic.response import text
@app.route("/cookie")
async def test(request):
response = text("There's a cookie up in this response")
response.cookies['test'] = 'It worked!'
response.cookies['test']['domain'] = '.gotta-go-fast.com'
response.cookies['test']['httponly'] = True
return response
10.3 删除Cookies
可以语义或明确地删除Cookies。
from sanic.response import text
@app.route("/cookie")
async def test(request):
response = text("Time to eat some cookies muahaha")
# This cookie will be set to expire in 0 seconds
del response.cookies['kill_me']
# This cookie will self destruct in 5 seconds
response.cookies['short_life'] = 'Glad to be here'
response.cookies['short_life']['max-age'] = 5
del response.cookies['favorite_color']
# This cookie will remain unchanged
response.cookies['favorite_color'] = 'blue'
response.cookies['favorite_color'] = 'pink'
del response.cookies['favorite_color']
return response
响应的cookies可以设置为字典值，同时也有以下参数可用：
expires(时间): cookie最后在客户端浏览器上存在时间。
path(字符串): Cookie的URL子集。默认为/。
comment(字符串): 注释(元数据)。
domain(字符串): 指定cookie有效的域。显式指定的域必须始终以点开头。
max-age(数字): cookie应该存在的秒数。
secure(布尔值): 指定cookie是否只能通过HTTPS发送。
httponly(布尔值): 指定cookie是否能被Javascript读取。
11. 处理器装饰
由于Sanic处理程序大都是简单的Python函数，因而你可以用类似Flask的方式对其进行装饰。典型的例子就是在你的执行程序之前运行一些你想运行的代码。
11.1 装饰授权
假设你需要检查用户是否有权访问特定的端点，你可以创建一个包装处理函数的装饰起，如果客户端有权访问资源，则检查请求，并发送适当的响应。
from functools import wraps
from sanic.response import json
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'})
12. 流媒体
12.1 请求流媒体
如下所示，Sanic允许你以流的方式请求数据。当请求结束的时候，``request.stream.get()将返回None`值。只有post、put和patch装饰器有流的参数。
from sanic import Sanic
from sanic.views import CompositionView
from sanic.views import HTTPMethodView
from sanic.views import stream as stream_decorator
from sanic.blueprints import Blueprint
from sanic.response import stream, text
bp = Blueprint('blueprint_request_stream')
app = Sanic('request_stream')
class SimpleView(HTTPMethodView):
@stream_decorator
async def post(self, request):
result = ''
while True:
body = await request.stream.get()
if body is None:
break
result += body.decode('utf-8')
return text(result)
@app.post('/stream', stream=True)
async def handler(request):
async def streaming(response):
while True:
body = await request.stream.get()
if body is None:
break
body = body.decode('utf-8').replace('1', 'A')
response.write(body)
return stream(streaming)
@bp.put('/bp_stream', stream=True)
async def bp_handler(request):
result = ''
while True:
body = await request.stream.get()
if body is None:
break
result += body.decode('utf-8').replace('1', 'A')
return text(result)
async def post_handler(request):
result = ''
while True:
body = await request.stream.get()
if body is None:
break
result += body.decode('utf-8')
return text(result)
app.blueprint(bp)
app.add_route(SimpleView.as_view(), '/method_view')
view = CompositionView()
view.add(['POST'], post_handler, stream=True)
app.add_route(view, '/composition_view')
if __name__ == '__main__':
app.run(host='127.0.0.1', port=8000)
12.2 响应流媒体
Sanic允许你使用流的方法将内容以流的方式传输到客户端。当传递一个可以写入的StreamingHTTPResponse对象时，这个方法接受协程回调(coroutine callback)。一个简单的例子如下：
from sanic import Sanic
from sanic.response import stream
app = Sanic(__name__)
@app.route("/")
async def test(request):
async def sample_streaming_fn(response):
response.write('foo,')
response.write('bar')
return stream(sample_streaming_fn, content_type='text/csv')
在你想将以流的方式传递内容到外部服务的客户端(如数据库)的时候很有用。例如，你可以使用asyncpg提供的异步游标将数据库的记录以流的方式传递到客户端。
@app.route("/")
async def index(request):
async def stream_from_db(response):
conn = await asyncpg.connect(database='test')
async with conn.transaction():
async for record in conn.cursor('SELECT generate_series(0, 10)'):
response.write(record[0])
return stream(stream_from_db)
13. 基于类的视图
基于类的视图只是为了实现对响应行为的请求的简单类。它们提供了在同一端点对不同HTTP请求类型进行区分处理的方法。
端点可以分配一个基于类的视图，而不是定义和装饰三种不同的处理函数和一个用于每个端点的请求类型。
13.1 定义视图
基于类的视图是HTTPMethodView的子类。你可以为每个HTTP请求实现你想要的类方法。如果一个请求没有定义方法，一个405:Method not allowed的响应就会生成。
要在端点上注册基于类的视图，就需要使用app.add_route方法。它的第一个参数是as_view方法定义的类，第二个参数是URL端点。
可被使用的方法包括get,post,patch,put和delete方法。使用这些方法的方式如下所示：
from sanic import Sanic
from sanic.views import HTTPMethodView
from sanic.response import text
app = Sanic('some_name')
class SimpleView(HTTPMethodView):
def get(self, request):
return text('I am get method')
def post(self, request):
return text('I am post method')
def put(self, request):
return text('I am put method')
def patch(self, request):
return text('I am patch method')
def delete(self, request):
return text('I am delete method')
app.add_route(SimpleView.as_view(), '/')
你也可以使用异步async语法。
from sanic import Sanic
from sanic.views import HTTPMethodView
from sanic.response import text
app = Sanic('some_name')
class SimpleAsyncView(HTTPMethodView):
async def get(self, request):
return text('I am async get method')
app.add_route(SimpleAsyncView.as_view(), '/')
13.2 URL 参数
如果你需要任何URL参数，就像路由章节介绍的一样，将其包含在方法定义中。
class NameView(HTTPMethodView):
def get(self, request, name):
return text('Hello {}'.format(name))
app.add_route(NameView.as_view(), '/')
13.3 装饰
如果你想添加任何装饰器到类中，可以设置decorators类变量。当调用as_view方法的时候，会应用于类中。
class ViewWithDecorator(HTTPMethodView):
decorators = [some_decorator_here]
def get(self, request, name):
return text('Hello I have a decorator')
app.add_route(ViewWithDecorator.as_view(), '/url')
13.4 构造URL
如果你希望为HTTPMethodView构造一个URL，类的名字将会当作url_for的端点。如下所示：
@app.route('/')
def index(request):
url = app.url_for('SpecialClassView')
return redirect(url)
class SpecialClassView(HTTPMethodView):
def get(self, request):
return text('Hello from the Special Class View!')
app.add_route(SpecialClassView.as_view(), '/special_class_view')
13.5 使用组成视图(CompositionView)
做为HTTPMethodView的替代方法，你可以在视图类外使用CompositionView来移动处理函数。
处理函数在来源的每个HTTP方法中都进行了定义，然后使用CompositionView.add方法来添加视图。
它的第一个参数应该是一个HTTP处理方法的列表(如['GET', 'POST'])，第二个参数是处理函数。
下面的例子展示了如何在CompositionView中使用外部处理函数和内联lanbda方法。
from sanic import Sanic
from sanic.views import CompositionView
from sanic.response import text
app = Sanic(__name__)
def get_handler(request):
return text('I am a get method')
view = CompositionView()
view.add(['GET'], get_handler)
view.add(['POST', 'PUT'], lambda request: text('I am a post/put method'))
# Use the new view to handle requests to the base URL
app.add_route(view, '/')
需要注意的是，当前你无法使用url_for方法为CompositionView构建一个URL。
14. 定制协议
你可以通过自定义一个协议来更改Sanic已经定义的协议行为，这个协议是asyncio.protocol的子类。这个协议能够以protocol关键字参数传递给sanic.run方法。
自定义协议类的构造函数接受以下的关键字参数：
loop: asyncio兼容的事件循环；
connections: 存储协议的set对象。当Sanic接收SIGINT和SIGTERM参数时，它会对集合中的所有协议对象执行protocol.close_if_idle。
signal: 一个具有stop属性的sanic.server.Signal对象。当Sanic接收到SIGINT和SIGTERM参数的时候，signal.stopped将被赋值为True。
request_handler: 一个将sanic.request.Request对象和response回调为参数的协程程序。
error_handler: 一个当出现异常时被调用出的sanic.exceptions.Handler对象。
request_timeout: 请求超时前的秒数。
request_max_size: 指定最大的请求数，以字节为单位。
14.1 例子
如果处理函数没有返回HTTPResponse对象，那没在默认的协议中就会发生错误。
通过复写write_response协议方法，如果处理程序返回一个字符串，它将被转换为一个HTTPResponse对象。
from sanic import Sanic
from sanic.server import HttpProtocol
from sanic.response import text
app = Sanic(__name__)
class CustomHttpProtocol(HttpProtocol):
def __init__(self, *, loop, request_handler, error_handler,
signal, connections, request_timeout, request_max_size):
super().__init__(
loop=loop, request_handler=request_handler,
error_handler=error_handler, signal=signal,
connections=connections, request_timeout=request_timeout,
request_max_size=request_max_size)
def write_response(self, response):
if isinstance(response, str):
response = text(response)
self.transport.write(
response.output(self.request.version)
)
self.transport.close()
@app.route('/')
async def string(request):
return 'string'
@app.route('/1')
async def response(request):
return text('response')
app.run(host='0.0.0.0', port=8000, protocol=CustomHttpProtocol)
15. SSL范例
SSLContext可以选择进行传递。
import ssl
context = ssl.create_default_context(purpose=ssl.Purpose.CLIENT_AUTH)
context.load_cert_chain("/path/to/cert", keyfile="/path/to/keyfile")
app.run(host="0.0.0.0", port=8443, ssl=context)
你还可以将证书和密钥的位置做为自检进行传递。
ssl = {'cert': "/path/to/cert", 'key': "/path/to/keyfile"}
app.run(host="0.0.0.0", port=8443, ssl=ssl)
16. 日志记录
Sanic允许你使用python3 的logging API对请求不同类型日志进行记录(诸如访问记录，错误记录)。
16.1 快速教程
下面是一个使用默认设置的简单示例：
from sanic import Sanic
from sanic.config import LOGGING
# The default logging handlers are ['accessStream', 'errorStream']
# but we change it to use other handlers here for demo purpose
LOGGING['loggers']['network']['handlers'] = [
'accessSysLog', 'errorSysLog']
app = Sanic('test')
@app.route('/')
async def test(request):
return response.text('Hello World!')
if __name__ == "__main__":
app.run(log_config=LOGGING)
如果需要关闭日志记录，只需要分配log_config=None:
if __name__ == "__main__":
app.run(log_config=None)
这将在处理请求的时候跳过调用日志的功能，你可以加快在使用中的速度。
if __name__ == "__main__":
# disable internal messages
app.run(debug=False, log_config=None)
16.2 配置
默认情况下，使用sanic.config.LOGGING字典来设置log_config参数，下面是handlers默认配置中设置的默认值：
internal: 使用logging.StreamHandler 内部信息在控制台输出。
accessStream: 使用logging.StreamHandler 登录控制台的请求信息。
errorStream: 使用logging.StreamHandler 控制台的错误信息和追溯信息。
accessSysLog: 使用logging.handlers.SysLogHandler 记录到syslog的请求信息。
errorSysLog: 使用logging.handlers.SysLogHandler syslog的错误消息和追溯记录。
filters过滤:
accessFilter: 使用sanic.log.DefaultFilter 只允许DEBUG，INFO和NONE(0)级别的过滤器。
errorFilter: 使用sanic.log.DefaultFilter 只允许在WARNING，ERROR和CRITICAL级别的过滤器。
sanic中使用了两种loggers，如果要创建自己的日志记录配置，则必须对它们进行定义：
sanic: 记录内部信息。
network: 记录来自网络请求，以及请求中的任何信息。
16.3 日志格式
除了由python(asctime,levelname,message)提供的默认参数之外，Sanic还为accessFilter提供了网络记录器的其他参数：
host (str) request.ip
request (str) request.method + " " + request.url
status (int) response.status
byte (int) len(response.body)
默认访问日志格式为:
%(asctime)s - (%(name)s)[%(levelname)s][%(host)s]: %(request)s %(message)s %(status)d %(byte)d
17. 测试
Sanic端点可以使用test_client对象进行本地测试，test_client对象依赖与aiohttp库。
该test_client展示了如何在你的app使用get,post,put,delete,patch,head和options方法。一个简单的例子(使用pytest)如下：
# Import the Sanic app, usually created with Sanic(__name__)
from external_server import app
def test_index_returns_200():
request, response = app.test_client.get('/')
assert response.status == 200
def test_index_put_not_allowed():
request, response = app.test_client.put('/')
assert response.status == 405
在内部每调用一次test_client方法的时候，运行在127.0.01:42101的Sanic的app的测试请求将使用aiohttp执行测试。
test_client方法接受以下的参数和关键字参数：
uri(默认'/'): 表示要测试的URL字符串。
gather_request(默认True): 布尔值，用于确定该函数是否返回原始请求，如果设置为True，则返回值是一个元组(request, response)，如果为False则只返回响应。
server_kwargs(默认为{}): 在运行测试请求之前传递给app.run的附加参数。
debug(默认False): 布尔值，用户确定是否在调试模式下运行该服务。
函数还将(request_args)和(*request_kwargs)直接传递给aiohttp ClientSession进行请求。
例如，要向GET请求中提供数据，可以执行一下的操作：
def test_get_request_includes_data():
params = {'key1': 'value1', 'key2': 'value2'}
request, response = app.test_client.get('/', params=params)
assert request.args.get('key1') == 'value1'
还有提供数据到 JSON POST请求中：
def test_post_json_request_includes_data():
data = {'key1': 'value1', 'key2': 'value2'}
request, response = app.test_client.post('/', data=json.dumps(data))
assert request.json.get('key1') == 'value1'
17.1 pytest-sanic
pytest-sanic是一个pytest插件，它可以帮助您异步测试代码。
就像下面这样进行测试：
async def test_sanic_db_find_by_id(app):
"""
Let's assume that, in db we have,
{
"id": "123",
"name": "Kobe Bryant",
"team": "Lakers",
}
"""
doc = await app.db["players"].find_by_id("123")
assert doc.name == "Kobe Bryant"
assert doc.team == "Lakers"
pytest-sanic还提供了一些有用的设置，如loop，unused_port，test_server，test_client。
@pytest.yield_fixture
def app():
app = Sanic("test_sanic_app")
@app.route("/test_get", methods=['GET'])
async def test_get(request):
return response.json({"GET": True})
@app.route("/test_post", methods=['POST'])
async def test_post(request):
return response.json({"POST": True})
yield app
@pytest.fixture
def test_cli(loop, app, test_client):
return loop.run_until_complete(test_client(app, protocol=WebSocketProtocol))
#########
# Tests #
#########
async def test_fixture_test_client_get(test_cli):
"""
GET request
"""
resp = await test_cli.get('/test_get')
assert resp.status == 200
resp_json = await resp.json()
assert resp_json == {"GET": True}
async def test_fixture_test_client_post(test_cli):
"""
POST request
"""
resp = await test_cli.post('/test_post')
assert resp.status == 200
resp_json = await resp.json()
assert resp_json == {"POST": True}
18. 部署
内建的Web服务器简化了Sanic的部署。在定义了一个sanic.Sanic的实例后，在调用run方法可以使用以下关键字参数：
host(默认“127.0.0.1”): 服务器主机的地址。
port(默认8000): 服务器的端口。
debug(默认False): 启用调试(减慢服务器速度)。
ssl(默认None): 用于工作者SSL加密的SSLContext。
sock(默认None)：服务器接受连接的Socket。
worker(默认值1)：生成的工作进程数。
loop(默认None): asyncio兼容的事件循环。如果没有指定，Sanic会创建自己的事件循环。
protocol(默认HttpProtocol)：asyncio.protocol的子类。
18.1 进程
默认情况下，Sanic在主进程中只侦听一个CPU内核。要启动其它核心，只需指定run参数中进程的数量。
app.run(host='0.0.0.0', port=1337, workers=4)
Sanic将自动启动多个进程并在它们之间建立路由路径。建议进程数和CPU核心数一样。
18.2 通过命令行运行
如果你喜欢使用命令行参数，则可以通过执行模块启动Sanic服务器。例如，如果你将Sanic应用程序在名为server.py的文件中初始化，那么可以像这样运行服务：
python -m sanic server.app --host=0.0.0.0 --port=1337 --workers=4
使用这种运行sanic的方法，没有必要在你的Python文件中调用app.run。如果需要这样做，请确保包装它，以便它由解释器直接执行。
if __name__ == '__main__':
app.run(host='0.0.0.0', port=1337, workers=4)
18.3 通过Gunicorn运行
Gunicorn'Green Unicorn'是用于UNIX的WSGI HTTP服务。
使用Gunicorn运行Sanic应用程序，您需要使用特殊的sanic.worker.GunicornWorker对象定义Gunicornworker-class参数：
gunicorn myapp:app --bind 0.0.0.0:1337 --worker-class sanic.worker.GunicornWorker
如果您的应用程序遇到内存泄漏，您可以配置Gunicorn在处理给定数量的请求后，正常地重新启动一个工作。这可以帮助限制内存泄漏的影响。
18.4 异步支持
异步支持合适与其他应用程序(特别是loop)共享sanic进程。但是请注意，因为此方法不支持使用多个进程，一般不是运行应用程序的首选方式。
下面是一个不完整的例子：
server = app.create_server(host="0.0.0.0", port=8000)
loop = asyncio.get_event_loop()
task = asyncio.ensure_future(server)
loop.run_forever()
19. 扩展
Sessions: session的支持，允许使用redis，memcache或内存进行存储。
CORS: 用于处理跨域资源共享的扩展。
Compress: 允许您轻松地压缩Sanic响应。
Jinja2: Jinja2模板框架。
OpenAPI/Swagger：OpenAPI支持，以及Swagger UI。
Pagination: 简单的分页支持。
Motor: Simple motor wrapper。
Sanic CRUD：基于peewee 模型的CRUD(创建/检索/更新/删除)REST API自动生成的框架。
UserAgent: 添加user_agent到请求
Limiter: 限制sanic速率。
Sanic EnvConfig：将环境变量加入sanic配置。
Babel：借助Babel库，向Sanic应用程序添加i18n/l10n支持。
Dispatch: 由werkzeug的DispatcherMiddleware驱动的调度程序。可以作为Sanic-to-WSGI适配器。
Sanic-OAuth: 用于连接和创建自己的token授权的库。
Sanic-nginx-docker-example: 在nginx使用docker-compose的一个简单易用的Sanic例子。
sanic-graphql: Sanic的GraphQL集成。
sanic-prometheus: Sanic的Prometheus指标。
Sanic-RestPlus: Sanic的Flask-RestPlus端口。基于SwaggerUI的全功能REST API。
sanic-transmute: 可从python函数和类生成API，并自动生成Swagger UI文档。
pytest-sanic: 一个用于Sanic的pytest插件。可以测试异步代码。
jinja2-sanic：一个用于Sanic的jinja2模板渲染器。
20. API参考