typora-copy-images-to: ipichtml
[TOC]node
在安裝Sanic以前,讓咱們一塊兒來看看Python在支持異步的過程當中,都經歷了哪些比較重大的更新。python
首先是Python3.4版本引入了asyncio
,這讓Python有了支持異步IO的標準庫,然後3.5版本又提供了兩個新的關鍵字async/await
,目的是爲了更好地標識異步IO,讓異步編程看起來更加友好,最後3.6版本更進一步,推出了穩定版的asyncio
,從這一系列的更新能夠看出,Python社區正邁着堅決且穩重的步伐向異步編程靠近。mysql
Sanic是一個支持 async/await
語法的異步無阻塞框架,這意味着咱們能夠依靠其處理異步請求的新特性來提高服務性能,若是你有Flask
框架的使用經驗,那麼你能夠迅速地使用Sanic
來構建出心中想要的應用,而且性能會提高很多,我將同一服務分別用Flask和Sanic編寫,再將壓測的結果進行對比,發現Sanic編寫的服務大概是Falsk
的1.5倍。nginx
僅僅是Sanic的異步特性就讓它的速度獲得這麼大的提高麼?是的,但這個答案並不標準,更爲關鍵的是Sanic使用了uvloop
做爲asyncio
的事件循環,uvloop
由Cython編寫,它的出現讓asyncio
更快,快到什麼程度?這篇文章中有介紹,其中提出速度至少比 nodejs、gevent 和其餘Python異步框架要快兩倍,而且性能接近於用Go編寫的程序,順便一提,Sanic的做者就是受這篇文章影響,這纔有了Sanic。git
怎麼樣?有沒有激起你學習Sanic的興趣,若是有,就讓咱們一塊兒開始學習吧,在開始以前,你只須要有一臺安裝了Python的電腦便可。github
說明:因爲Windows下暫不支持安裝uvloop,故在此建議使用Mac或Linux
程序世界一部分是對應着現實的,在生活中,咱們會在不一樣的環境完成不一樣的任務,好比在廚房作飯、臥室休息,分工極其明確。web
其實用Python編寫應用服務也是如此,它們一樣但願應用服務與開發環境是一對一的關係,這樣作的好處在於,每一個獨立的環境均可以簡潔高效地管理自身對應服務所依賴的第三方庫,如若否則,各個服務都安排在同一環境,這樣不只會形成管理上的麻煩,還會使第三方庫之間產生衝突。redis
經過上面的敘述,咱們是否是能夠得出這樣一個核心觀點:應該在不一樣的環境下作不一樣的事 ,以此類推,寫項目的時候,咱們也須要爲每一個不一樣的項目構建一個無干擾的的環境,發散思惟,總結一下:sql
不一樣的項目,須要爲其構建不一樣的虛擬環境,以避免互相干擾
構建虛擬環境的工具不少,以下:
…...
以上三個工具均可以快速地幫助咱們構建當前須要的Python環境,若是你以前沒有使用過,可直接點開連接進行下載,若是你正在使用其它的環境管理工具,也沒關係,由於不論你使用哪種方式,咱們最終目的都是針對一個新項目構建一個新的環境。
安裝配置好以後,簡單看看官方提供的使用方法,就能夠開始了,好比我本機使用的是venv
(python3.5之後官方推薦使用這個venv來管理虛擬環境),安裝完成後能夠很方便地建立一個虛擬環境,好比這裏使用Python3.6來做爲本書項目的默認環境:
cd ~/ # 新建一個python3.6環境 python3 -m venv pyenv # 安裝好以後 輸入下面命令進入名爲python36的環境 cd pyenv/ source bin/activate # 查看版本 python -V
若安裝速度比較慢,能夠考慮換國內源,好比 國內鏡像 ,至於爲何選擇python3.6做爲默認環境,一是由於Sanic只支持Python3.5+,二則是咱們構建的項目最終是要在生產環境下運行的,因此建議最好安裝Python3.6下穩定版本的asyncio
。
Python安裝第三方模塊都是利用pip
工具進行安裝,這裏也不例外,首先進入上一步咱們新建的 python3.6
虛擬環境,而後安裝:
# 安裝Sanic,請先使用 source activate python36 進入虛擬環境 pip install sanic # 若是不想使用uvloop和ujson 能夠這樣安裝 SANIC_NO_UVLOOP=true SANIC_NO_UJSON=true pip install sanic
經過上面的命令,你就能夠在 python3.6
虛擬環境中安裝Sanic以及其依賴的第三方庫了,若想查看Sanic是否已經正確安裝,能夠進入終端下對應的虛擬環境,啓動Python解釋器,導入Sanic庫:
python >>> >>> import sanic
若是沒有出現錯誤,就說明你已經正確地安裝了Sanic,請繼續閱讀下一節,瞭解下如何利用Sanic來構建一個Web項目吧。
咱們將正式使用Sanic來構建一個web項目,讓咱們踏出第一步,利用Sanic來編寫一個返回Hello World!
字符串的服務程序。
新建一個文件夾sanicweb
:
$ mkdir sanicweb $ cd sanicweb/ $ pwd /Users/junxi/pyenv/sanicweb
建立一個sanic例子,保存爲 main.py
:
from sanic import Sanic from sanic.response import text app = Sanic() @app.route("/") async def index(request): return text('Hello World!') if __name__ == "__main__": app.run(host="0.0.0.0", port=9000)
運行main.py
,而後訪問地址http://127.0.0.1:9000/
$ curl -X GET http://127.0.0.1:9000/ Hello World!
這樣咱們就完成了第一個sanic例子。
接下來,你將逐漸地瞭解到Sanic的一些基本用法,如路由的構建、接受請求數據以及返回響應的內容等。
路由容許用戶爲不一樣的URL端點指定處理程序函數。
實例:
from sanic.response import json @app.route("/") async def index(request): return json({ "hello": "world" })
url http://server.url/ 被訪問(服務器的基本url),最終/被路由器匹配處處理程序函數,測試,而後返回一個JSON對象。
必須使用async def語法來定義Sanic處理函數,由於它們是異步函數。
要指定一個參數,能夠用像這樣的角引號<PARAM>包圍它。請求參數將做爲關鍵字參數傳遞給路線處理程序函數。
實例:
@app.router('/tag/<tag>') async def tag_handler(request, tag): return text('Tag - {}'.format(tag))
重啓服務,輸入地址http://127.0.0.1:9000/tag/python
進行訪問
$ curl -X GET http://127.0.0.1:9000/tag/python Tag - python
爲參數指定類型,在參數名後面添加(:類型)。若是參數不匹配指定的類型,Sanic將拋出一個不存在的異常,致使一個404頁面
@app.route('/number/<integer_arg:int>') async def integer_handler(request, integer_arg): return text('Integer - {}'.format(integer_arg)) @app.route('/number/<number_arg:number>') async def number_handler(request, number_arg): return text('Number - {}'.format(number_arg)) @app.route('/person/<name:[A-z]+>') async def person_handler(request, name): return text('Person - {}'.format(name)) @app.route('/folder/<folder_id:[A-z0-9]{0,4}>') async def folder_handler(request, folder_id): return text('Folder - {}'.format(folder_id))
測試結果以下:
$ curl -X GET http://127.0.0.1:9000/number/1 Integer - 1 $ curl -X GET http://127.0.0.1:9000/number/asds Error: Requested URL /number/asds not found $ curl -X GET http://127.0.0.1:9000/number/12.0 Number - 12.0 $ curl -X GET http://127.0.0.1:9000/person/junxi Person - junxi $ curl -X GET http://127.0.0.1:9000/person/123 Error: Requested URL /person/123 not found $ curl -X GET http://127.0.0.1:9000/folder/img Folder - img $ curl -X GET http://127.0.0.1:9000/folder/img1 Folder - img1 $ curl -X GET http://127.0.0.1:9000/folder/images Error: Requested URL /folder/images not found $ curl -X GET http://127.0.0.1:9000/folder/2018 Folder - 2018
路由裝飾器接受一個可選的參數,方法,它容許處理程序函數與列表中的任何HTTP方法一塊兒工做。
實例1:
@app.route('/post1', methods=['POST']) async def post_handler(request): return text('POST request - {}'.format(request.json)) @app.route('/get1', methods=['GET']) async def get_handler(request): return text('GET request - {}'.format(request.args))
實例2:
@app.post('/post2') async def post_handler(request): return text('POST request - {}'.format(request.json)) @app.get('/get2') async def get_handler(request): return text('GET request - {}'.format(request.args))
測試結果:
$ curl -X GET http://127.0.0.1:9000/get1?name=junxi GET request - {'name': ['junxi']} $ curl -X GET http://127.0.0.1:9000/get2?name=junxi GET request - {'name': ['junxi']} $ curl -H "Content-type: application/json" -X POST -d '{"name":"junxi", "gender":"male"}' http://127.0.0.1:9000/post1 POST request - {'name': 'junxi', 'gender': 'male'} $ curl -H "Content-type: application/json" -X POST -d '{"name":"junxi", "gender":"male"}' http://127.0.0.1:9000/post2 POST request - {'name': 'junxi', 'gender': 'male'}
實例:
async def handler1(request): return text('ok') async def handler2(request, name): return text('Folder - {}'.format(name)) async def personal_handler2(request, name): return text('Person - {}'.format(name)) app.add_route(handler1, '/test1') app.add_route(handler2, '/folder2/<name>') app.add_route(personal_handler2, '/personal2/<name:[A-z]>', methods=['GET'])
測試結果:
$ curl -X GET http://127.0.0.1:9000/test1 ok $ curl -X GET http://127.0.0.1:9000/folder2/aaa Folder - aaa $ curl -X GET http://127.0.0.1:9000/personal2/A Person - A $ curl -X GET http://127.0.0.1:9000/personal2/a Person - a
Sanic提供了一個urlfor方法,根據處理程序方法名生成url。避免硬編碼url路徑到您的應用程序
實例:
@app.router("/") async def index(request): url = app.url_for('post_handler', post_id=5) return redirect(url) @app.route('posts/<post_id>') async def post_handler(request, post_id): return text('Post - {}'.format(post_id))
給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以便構建一個URL。若是沒有提供一個參數,或者一個參數與指定的類型不匹配,就會拋出一個URLBuildError
能夠將多值參數傳遞給url
url = app.url_for('post_handler', post_id=5, arg_one=['one', 'two']) # /posts/5?arg_one=one&arg_one=two
通過測試訪問/
咱們會發現,url跳轉到了/posts/5
,並打印出來Post - 5 的結果。
redirect
是從sanic.response
導入的方法,用於處理url的重定向。
WebSocket routes
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)
另外,添加websocket路由方法能夠代替裝飾器
async def feed(request, ws): pass app.add_websocket_route(my_websocket_handler, '/feed')
request
當一個端點收到一個HTTP請求時,路由功能被傳遞給一個 Request
對象。
如下變量可做爲Request
對象的屬性訪問:
json
(any) - JSON bodyfrom sanic.response import json @app.route("/json") def post_json(request): return json({ "received": True, "message": request.json })
args
(dict) - 查詢字符串變量。查詢字符串是相似於URL的部分?key1=value1&key2=value2
。若是該URL被解析,則args
字典將以下所示:{'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
(dictionary of File
objects) - 具備名稱,正文和類型的文件列表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) - post表單變量。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
(bytes) - 發送原始主體。不管內容類型如何,該屬性都容許檢索請求的原始數據。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)
headers
(dict) - 包含請求標頭的不區分大小寫的字典。ip
(str) - 請求者的IP地址。port
(str) - 請求者的端口地址。socket
(tuple) - 請求者的(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
:請求的路徑: /posts/1/
query_string
:請求的查詢字符串:foo=bar
或一個空白字符串''
uri_template
:匹配路由處理程序的模板: /posts/<id>/
token
:受權標頭(Authorization)的值: Basic YWRtaW46YWRtaW4=
get
和getlist
訪問數據返回字典的請求屬性實際上會返回一個dict
被調用的子類 RequestParameters
。使用這個對象的關鍵區別在於get
和getlist
方法之間的區別。
get(key, default=None)
按照正常操做,除了當給定鍵的值是列表時,只返回第一個項目。getlist(key, default=None)
正常操做,返回整個列表。response
from sanic import response @app.route('/text') def handle_request(request): return response.text('Hello world!')
from sanic import response @app.route('/html') def handle_request(request): return response.html('<p>Hello world!</p>')
from sanic import response @app.route('/json') def handle_request(request): return response.json({'message': 'Hello world!'})
from sanic import response @app.route('/file') async def handle_request(request): return await response.file('/srv/www/whatever.png')
流媒體
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')
對於大文件,文件和流的組合
from sanic import response @app.route('/big_file.png') async def handle_request(request): return await response.file_stream('/srv/www/whatever.png')
from sanic import response @app.route('/redirect') def handle_request(request): return response.redirect('/json')
沒有進行編碼的響應
from sanic import response @app.route('/raw') def handle_request(request): return response.raw('raw data')
要修改頭部或狀態代碼,將頭部或狀態參數傳遞給這些函數
from sanic import response @app.route('/json') def handle_request(request): return response.json( {'message': 'Hello world!'}, headers={'X-Served-By': 'sanic'}, status=200 )
static_files
靜態文件和目錄,好比一個圖像文件,在Sanic註冊時使用。該方法使用一個端點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)
Note:目前,您不能使用url構建一個靜態文件的URL
html templates編寫
編寫web服務,天然會涉及到html,sanic自帶有html函數,但這並不能知足有些需求,故引入jinja2迫在眉睫。使用方法也很簡單:
# novels_blueprint.py片斷 from sanic import Blueprint from jinja2 import Environment, PackageLoader, select_autoescape # 初始化blueprint並定義靜態文件夾路徑 bp = Blueprint('novels_blueprint') bp.static('/static', './static/novels') # jinjia2 config env = Environment( loader=PackageLoader('views.novels_blueprint', '../templates/novels'), autoescape=select_autoescape(['html', 'xml', 'tpl'])) def template(tpl, **kwargs): template = env.get_template(tpl) return html(template.render(kwargs)) @bp.route("/") async def index(request): return template('index.html', title='index')
這樣,就實現了jinja2 模版的引入。
Exceptions
要拋出異常,只需從sanic異常模塊中提出相應的異常。
from sanic.exceptions import ServerError @app.route('/killme') def i_am_ready_to_die(request): raise ServerError("Something bad happened", status_code=500)
也能夠自定義狀態碼
from sanic.exceptions import abort from sanic.response import text @app.route('/youshallnotpass') def no_no(request): abort(401) # this won't happen text("OK")
Handling exceptions
裝飾器一個異常列表做爲參數來處理。你能夠經過SanicException來捕獲它們!裝飾異常處理函數必須將請求和異常對象做爲參數。
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)) @app.exception(NotFound) def handle_404_redirect(request, exception): uri = app.url_for('index') return redirect(uri)
Useful exceptions
經常使用
Middleware And Listeners
Middleware
有兩種類型的中間件: 請求和響應。二者都是使用@app
聲明的。中間件裝飾器,裝飾器的參數是一個表明其類型的字符串:「請求」或「響應」。響應中間件接收請求和響應做爲參數。
最簡單的中間件根本不修改請求或響應
@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")
中間件能夠修改它所提供的請求或響應參數,只要它不返回它
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)
上述代碼將按順序應用這兩個中間件。首先,中間件custombanner將把HTTP響應頭服務器更改成假服務器,而第二個中間件防止XSS將添加HTTP頭來防止跨站點腳本攻擊(XSS)攻擊。這兩個函數是在用戶函數返回響應以後調用的。
Listeners
若是想在服務器啓動或關閉時執行啓動/分解代碼,可使用如下偵聽器:
這些監聽器在函數中實現爲修飾符,它們接受應用程序對象和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()
若是你想在循環開始後運行一個後臺任務,那麼Sanic就提供了addtask方法來輕鬆地完成這一任務。
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())
Blueprints
藍圖是能夠用於應用程序中的子路由的對象。除了嚮應用程序實例添加路由,藍圖還定義了相似的添加路由的方法,而後以靈活的可插入的方式在應用程序中註冊。
simple Blueprint
假設將該文件保存爲myblueprint。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'})
registering blueprints
藍圖必須在應用程序中註冊
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)
Use_blueprint
WebSocket routes
WebSocket處理程序能夠註冊,使用@bp.websocket
裝飾或bp.add_websocket_route
方法
Middleware
使用藍圖還能夠在全局內註冊中間件。
@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')
Exception
異常狀況能夠用於全局的藍圖
@bp.exception(NotFound) def ignore_404s(request, exception): return text("Yep, I totally found the page: {}".format(request.url))
Static files
靜態文件能夠添加前綴
bp.static('/folder/to/serve', '/web/path')
藍圖能夠在服務器的啓動和中止過程當中運行函數。若是在多處理器模式下運行(超過1個worker),這些都是在workers fork以後觸發的。
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()
Use-case: API versioning
藍圖對於API版本控制是很是有用的,其中一個藍圖可能指向/v1/<route>,另外一個指向/v2/<route>。
當一個藍圖被初始化時,它能夠選擇一個可選的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)
若是但願在blueprint內部路由生成一個URL,記住,端點名稱採用格式<blueprint_name>.<handler_name>
@blueprint_v1.route('/') async def root(request): url = app.url_for('v1.post_handler', post_id=5) # --> '/v1/post/5' # url = request.app.url_for('v1.post_handler', post_id=5) # --> '/v1/post/5' return redirect(url) @blueprint_v1.route('/post/<post_id>') async def post_handler(request, post_id): return text('Post {} in Blueprint V1'.format(post_id))
Note: 當app和blueprint不在同一個模塊內記得加上request
例如:url = request.app.url_for('v1.post_handler', post_id=5) # --> '/v1/post/5'
Configuration
任何一個至關複雜的應用程序都須要配置,而不是在實際代碼中進行。對於不一樣的環境或安裝,設置多是不一樣的。
Sanic在應用程序對象的配置屬性中保持配置。配置對象僅僅是一個可使用點符號或字典來修改的對象。
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)
通常來講,該約定只具備大寫的配置參數。下面描述的加載配置的方法只會查找這些大寫的參數。
如何加載配置有幾種方法。
使用SANIC_前綴定義的任何變量都將應用於sanic config。例如,設置SANIC_REQUEST_TIMEOUT
將由應用程序自動加載並輸入REQUEST_TIMEOUT
配置變量。你能夠經過一個不一樣的前綴到Sanic:
app = Sanic(load_env='MYAPP_')
而後,上面的變量將是MYAPP_REQUEST_TIMEOUT。若是您想要禁用環境變量的加載,您能夠將其設置爲False:
app = Sanic(load_env=False)
若是有不少配置值,並且它們有合理的默認值,那麼將它們放到一個模塊中可能會有幫助:
import myapp.default_settings app = Sanic('myapp') app.config.from_object(myapp.default_settings)
您也可使用類或任何其餘對象。
一般,您須要從一個不屬於分佈式應用程序的文件中加載配置。可使用from_pyfile(/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文件,這些文件是爲了加載它們而執行的。這容許您使用任意邏輯來構造正確的配置。在配置中只添加了大寫的變量。最多見的配置包括簡單的鍵值對:
# config_file DB_HOST = 'localhost' DB_NAME = 'appdb' DB_USER = 'appuser'
在這個框中,只有幾個預約義值,在建立應用程序時能夠重寫。
| Variable | Default | Description | | ------------------ | --------- | --------------------------------------------- | | REQUEST_MAX_SIZE | 100000000 | How big a request may be (bytes) | | REQUEST_TIMEOUT | 60 | How long a request can take to arrive (sec) | | RESPONSE_TIMEOUT | 60 | How long a response can take to process (sec) | | KEEP_ALIVE | True | Disables keep-alive when False | | KEEP_ALIVE_TIMEOUT | 5 | How long to hold a TCP connection open (sec) |
請求超時度量在新打開的TCP鏈接被傳遞給Sanic
後端服務器時的時間間隔,以及接收整個HTTP請求的瞬間。若是時間超過了REQUEST_TIMEOUT
值(以秒爲單位),那麼這將被視爲客戶端錯誤,所以Sanic生成一個HTTP 408響應並將其發送給客戶端。若是您的客戶常常經過很是大的請求負載或者很是緩慢地上傳請求,請調整這個值。
響應超時度量在Sanic服務器將HTTP請求傳遞到Sanic應用程序的時間之間的時間,以及發送到客戶機的HTTP響應的時間。若是時間超過了RESPONSE_TIMEOUT
值(以秒爲單位),這被認爲是服務器錯誤,所以Sanic生成一個HTTP 503響應並將其設置爲客戶機。若是應用程序極可能長時間運行,延遲響應的生成,則將此值調整得更高。
Keep-Alive是HTTP 1.1中的一個HTTP特性。發送HTTP請求時,客戶端(一般是web瀏覽器應用程序)能夠設置一個keepalive消息頭,以指示HTTP服務器(Sanic)在發送響應以後不關閉TCP鏈接。這容許客戶端重用現有的TCP鏈接來發送後續的HTTP請求,並確保客戶機和服務器的網絡流量更高效。
在Sanic中,KEEP_ALIVE
配置變量默認設置爲True
。若是您在應用程序中不須要此功能,則將其設置爲False
,以便在發送響應後當即關閉全部客戶端鏈接,而不考慮請求上的keepalive消息頭。
服務器保持TCP鏈接打開的時間量由服務器自己決定。在Sanic中,該值使用KEEP_ALIVE_TIMEOUT
值進行配置。默認狀況下,它設置爲5秒,這是與Apache HTTP服務器相同的默認設置,在容許客戶端發送新請求的足夠時間和不一樣時打開太多鏈接之間保持良好的平衡。不要超過75秒,除非你知道你的客戶正在使用支持TCP鏈接的瀏覽器。
供參考:
Apache httpd server default keepalive timeout = 5 seconds Nginx server default keepalive timeout = 75 seconds Nginx performance tuning guidelines uses keepalive = 15 seconds IE (5-9) client hard keepalive limit = 60 seconds Firefox client hard keepalive limit = 115 seconds Opera 11 client hard keepalive limit = 120 seconds Chrome 13+ client keepalive limit > 300+ seconds
cookie是保存在用戶瀏覽器內的數據塊。Sanic既能夠讀寫cookie,也能夠存儲爲鍵-值對。
Warning
cookie能夠由客戶機自由修改。所以,您不能僅在cookie中存儲諸如登陸信息這樣的數據,由於客戶機能夠隨意更改這些數據。爲了確保存儲在cookie中的數據不會被客戶僞造或篡改, use something like itsdangerous to cryptographically sign the data.
用戶的cookie能夠經過請求對象的cookie字典訪問。
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))
返回響應時,能夠在響應對象上設置cookie。
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
cookie能夠經過語義或顯式刪除。
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
響應cookie能夠設置爲字典值,並具備如下參數:
sanic對此有一個第三方插件sanic_session
,用法很是簡單,見官方例子以下:
import asyncio_redis from sanic import Sanic from sanic.response import text from sanic_session import RedisSessionInterface app = Sanic() # Token from https://github.com/subyraman/sanic_session class Redis: """ A simple wrapper class that allows you to share a connection pool across your application. """ _pool = None async def get_redis_pool(self): if not self._pool: self._pool = await asyncio_redis.Pool.create( host='localhost', port=6379, poolsize=10 ) return self._pool redis = Redis() # pass the getter method for the connection pool into the session session_interface = RedisSessionInterface(redis.get_redis_pool, expiry=604800) @app.middleware('request') async def add_session_to_request(request): # before each request initialize a session # using the client's request await session_interface.open(request) @app.middleware('response') async def save_session(request, response): # after each request save the session, # pass the response to set client cookies await session_interface.save(request, response) @app.route("/") async def test(request): # interact with the session like a normal dict if not request['session'].get('foo'): request['session']['foo'] = 0 request['session']['foo'] += 1 response = text(request['session']['foo']) return response if __name__ == "__main__": app.run(host="0.0.0.0", port=8888, debug=True)
因爲Sanic處理程序是簡單的Python函數,您能夠用相似於Flask的方式向它們應用decorator。一個典型的用例是,當您須要一些代碼在處理程序的代碼執行以前運行。
假設您想要檢查用戶是否被受權訪問某個特定的端點。您能夠建立包裝處理函數的decorator,檢查客戶端是否被受權訪問某個資源,併發送適當的響應。
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'})
流媒體
Sanic容許您經過流獲取請求數據,以下所示。當請求結束時,request.stream.get()
返回None
。只有post, put和patch decorator 有流參數。
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)
Sanic容許您使用stream
方法將內容流到客戶機。該方法接受一個coroutine回調,該回調將傳遞給寫入的StreamingHTTPResponse
對象。一個簡單的例子以下:
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)
基於類的視圖只是實現對請求的響應行爲的類。它們提供了一種方法,將不一樣的HTTP請求類型劃分到同一端點。與其定義和修飾三個不一樣的處理函數(每一個端點支持的請求類型),端點能夠分配一個基於類的視圖。
基於類的視圖應該子類化HTTPMethodView
。而後,您能夠爲但願支持的每一個HTTP請求類型實現類方法。若是接收到的請求沒有定義的方法,則會生成一個405: Method not allowed 的響應。
要在端點上註冊基於類的視圖,將使用app.add_route
方法。第一個參數應該是被調用的方法as_view
的定義類,第二個參數應該是URL端點。
可用的方法是get
、post
、put
、patch
和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(), '/')
若是您須要任何URL參數,如路由指南中所討論的,在方法定義中包含它們。
class NameView(HTTPMethodView): def get(self, request, name): return text('Hello {}'.format(name)) app.add_route(NameView.as_view(), '/<name>')
若是您想在類中添加任何裝飾器,能夠設置decorator
類變量。當調用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')
若是您但願爲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')
Using CompositionView
做爲HTTPMethodView
的替代方法,您可使用CompositionView
將處理程序函數移到視圖類以外。
每一個支持的HTTP方法的處理函數都在源代碼的其餘地方定義,而後使用CompositionView.add
方法添加到視圖中。第一個參數是要處理的HTTP方法的列表(例如,['GET', 'POST']
),第二個參數是處理函數。下面的示例顯示了使用外部處理程序函數和內聯lambda的CompositionView
用法:
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, '/')
Note: 當前您不能使用url_for
爲CompositionView構建一個URL。
注意:這是高級用法,大多數讀者不須要這樣的功能。
您能夠經過指定自定義協議來更改Sanic協議的行爲,該協議應該是asyncio.protocol的子類。而後,該協議能夠做爲sanic.run
方法的關鍵字參數協議傳遞。
自定義協議類的構造函數接收來自Sanic的如下關鍵字參數。
loop
: 一個異步兼容的事件循環。connections
: 用於存儲協議對象的集合。當Sanic接收SIGINT
或SIGTERM
時,它執行protocol.close_if_idle
關閉此集合中存儲的全部協議對象。signal
: 帶有stopped
屬性的sanic.server.Signal
對象。當Sanic收到SIGINT
或SIGTERM
,signal.stopped
分配True
。request_handler
: 取一個sanic.request.Request
對象和response
回調做爲參數的coroutine。error_handler
: 在拋出異常時調用的處理程序sanic.exceptions.Handler
。request_timeout
: 請求超時前的秒數。request_max_size
: 指定請求的最大大小的整數,以字節爲單位。若是處理函數不返回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)
能夠傳入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)
Logging
Sanic容許您根據python3 logging API對請求進行不一樣類型的日誌記錄(訪問日誌、錯誤日誌)。若是您想建立一個新的配置,您應該對python3 logging有一些基本的瞭解。
使用默認設置的一個簡單示例以下:
from sanic import Sanic app = Sanic('test') @app.route('/') async def test(request): return response.text('Hello World!') if __name__ == "__main__": app.run(debug=True, access_log=True)
要使用本身的日誌記錄配置,只需使用logging.config.dictConfig
,或在初始化Sanic
應用時傳遞log_confi
g:
app = Sanic('test', log_config=LOGGING_CONFIG)
要關閉日誌,只需分配access_log=False:
if __name__ == "__main__": app.run(access_log=False)
這將跳過在處理請求時調用日誌功能。你甚至能夠作進一步的生產以得到額外的速度:
if __name__ == "__main__": # disable debug messages app.run(debug=False, access_log=False)
默認狀況下,log_config
參數設置爲使用sanic.log.LOGGING_CONFIG_DEFAULTS
字典配置。
在sanic中使用了三個日誌記錄器loggers
,若是您想建立本身的日誌配置,則必須定義:
root: 用於記錄內部消息。sanic.error: 用於記錄錯誤日誌。
sanic.access: 用於記錄訪問日誌。
除了python提供的默認參數(asctime、levelname、message), Sanic還提供了用於訪問日誌記錄器logger
的其餘參數:
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
Testing
Sanic端點可使用test_client對象在本地測試,這取決於附加的aiohttp庫。
test_client
公開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
方法時,Sanic應用程序運行於127.0.0.1:42101
,您的測試請求使用aiohttp
執行。
test_client
方法接受如下參數和關鍵字參數:
uri
(default'/'
) 一個表示測試uri的字符串。
gather_request
(defaultTrue
) 一個布爾值,它決定原始請求是否由函數返回。若是設置爲True
,返回值是(request, response)
的一個元組,若是False
僅返回響應。
server_kwargs
(default{}
) 在測試請求運行以前傳遞給app.run
的附加參數。
debug
(defaultFalse
)一個布爾值,它決定是否在調試模式下運行服務器。
該函數進一步接受了*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'
關於aiohttp的可用參數的更多信息能夠在ClientSession的文檔中找到。
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}
Deploying
部署Sanic是由內置的webserver簡化的。在定義了sanic.Sanic
的實例以後。咱們能夠用如下關鍵字參數調用run
方法:
host
(default "127.0.0.1"
): 地址來託管服務器。port
(default 8000
): 開啓服務器的端口。debug
(default False
): 啓用調試輸出(減慢服務器)。ssl
(default None
): ssl加密的SSLContext。sock
(default None
): 用於服務器接受鏈接的套接字。workers
(default 1
): 生成的工做進程數。loop
(default None
): 一個asyncio兼容的事件循環。若是沒有指定,Sanic將建立本身的事件循環。protocol
(default HttpProtocol
): asyncio.protocol的子類。Workers
默認狀況下,Sanic只使用一個CPU核心偵聽主進程。To crank up the juice,只需在run
參數中指定workers的數量。
app.run(host='0.0.0.0', port=1337, workers=4)
Sanic將會自動啓動多個進程,並在它們之間路由流量。咱們建議儘量多的workers擁有可用的核心。
若是您喜歡使用命令行參數,則能夠經過執行模塊來啓動Sanic服務器。例如,若是您在名爲server.py
的文件中初始化Sanic做爲app
,你能夠這樣運行服務器:
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)
Gunicorn ‘Green Unicorn’ 是UNIX的一個WSGI HTTP服務器。它是一個由Ruby的Unicorn項目移植的預fork worker模型。
爲了使用Gunicorn運行Sanic應用程序,您須要爲Gunicornworker-class
參數使用特殊的sanic.worker.GunicornWorker
:
gunicorn myapp:app --bind 0.0.0.0:1337 --worker-class sanic.worker.GunicornWorker
若是您的應用程序遭受內存泄漏,您能夠配置Gunicorn以優雅地從新啓動一個worker,由於它處理了給定數量的請求。這能夠方便地幫助限制內存泄漏的影響。
有關更多信息,請參見 Gunicorn 文檔。
Asynchronous support
若是您須要與其餘應用程序共享sanic進程,特別是loop
,這是合適的。可是,請注意,該方法不支持使用多進程,而且不是通常運行該應用程序的首選方式。
這裏有一個不完整的示例(請參見run_asyn.py
在一些更實用的例子中):
server = app.create_server(host="0.0.0.0", port=8000) loop = asyncio.get_event_loop() task = asyncio.ensure_future(server) loop.run_forever()
Extensions
由社區建立的Sanic擴展列表。
user_agent
to request。Babel
庫的幫助下向Sanic應用程序添加i18n/l10n支持。DispatcherMiddleware
激發的調度程序。能夠充當sanicto - wsgi適配器。