sanic非同步框架之中文檔案

阿新 • • 發佈：2019-12-31

typora-copy-images-to: ipic

[TOC]

快速開始

在安裝Sanic之前，讓我們一起來看看Python在支援非同步的過程中，都經歷了哪些比較重大的更新。

首先是Python3.4版本引入了asyncio，這讓Python有了支援非同步IO的標準庫，而後3.5版本又提供了兩個新的關鍵字async/await，目的是為了更好地標識非同步IO，讓非同步程式設計看起來更加友好，最後3.6版本更進一步，推出了穩定版的asyncio，從這一系列的更新可以看出，Python社群正邁著堅定且穩重的步伐向非同步程式設計靠近。

安裝

Sanic是一個支援 async/await 語法的非同步無阻塞框架，這意味著我們可以依靠其處理非同步請求的新特性來提升服務效能，如果你有Flask

框架的使用經驗，那麼你可以迅速地使用Sanic來構建出心中想要的應用，並且效能會提升不少，我將同一服務分別用Flask和Sanic編寫，再將壓測的結果進行對比，發現Sanic編寫的服務大概是Falsk的1.5倍。

僅僅是Sanic的非同步特性就讓它的速度得到這麼大的提升麼？是的，但這個答案並不標準，更為關鍵的是Sanic使用了uvloop作為asyncio的事件迴圈，uvloop由Cython編寫，它的出現讓asyncio更快，快到什麼程度？這篇文章中有介紹，其中提出速度至少比 nodejs、gevent 和其他Python非同步框架要快兩倍，並且效能接近於用Go編寫的程式，順便一提，Sanic的作者就是受這篇文章影響，這才有了Sanic。

怎麼樣？有沒有激起你學習Sanic的興趣，如果有，就讓我們一起開始學習吧，在開始之前，你只需要有一臺安裝了Python的電腦即可。

說明：由於Windows下暫不支援安裝uvloop，故在此建議使用Mac或Linux

虛擬環境

程式世界一部分是對應著現實的，在生活中，我們會在不同的環境完成不同的任務，比如在廚房做飯、臥室休息，分工極其明確。

其實用Python編寫應用服務也是如此，它們同樣希望應用服務與開發環境是一對一的關係，這樣做的好處在於，每個獨立的環境都可以簡潔高效地管理自身對應服務所依賴的第三方庫，如若不然，各個服務都安排在同一環境，這樣不僅會造成管理上的麻煩，還會使第三方庫之間產生衝突。

通過上面的敘述，我們是不是可以得出這樣一個核心觀點：應該在不同的環境下做不同的事 ，以此類推，寫專案的時候，我們也需要為每個不同的專案構建一個無幹擾的的環境，發散思維，總結一下：

不同的專案，需要為其構建不同的虛擬環境，以免互相干擾

構建虛擬環境的工具很多，如下：

…...

以上三個工具都可以快速地幫助我們構建當前需要的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。

安裝Sanic

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 server.url/ 被訪問(伺服器的基本url)，最終/被路由器匹配到處理程式函式，測試，然後返回一個JSON物件。

必須使用async def語法來定義Sanic處理函式，因為它們是非同步函式。

請求引數

要指定一個引數，可以用像這樣的角引號包圍它。請求引數將作為關鍵字引數傳遞給路線處理程式函式。

例項：

@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
複製程式碼

url_for

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',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 body

from 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

text

from sanic import response

@app.route('/text')
def handle_request(request):
    return response.text('Hello world!')
複製程式碼

HTML

from sanic import response

@app.route('/html')
def handle_request(request):
    return response.html('<p>Hello world!</p>')
複製程式碼

JSON

from sanic import response

@app.route('/json')
def handle_request(request):
    return response.json({'message': 'Hello world!'})
複製程式碼

File

from sanic import response

@app.route('/file')
async def handle_request(request):
    return await response.file('/srv/www/whatever.png')
複製程式碼

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')
複製程式碼

File Streaming

對於大檔案，檔案和流的組合

from sanic import response

@app.route('/big_file.png')
async def handle_request(request):
    return await response.file_stream('/srv/www/whatever.png')
複製程式碼

Redirect

from sanic import response

@app.route('/redirect')
def handle_request(request):
    return response.redirect('/json')
複製程式碼

Raw

沒有進行編碼的響應

from sanic import response

@app.route('/raw')
def handle_request(request):
    return response.raw('raw data')
複製程式碼

Modify headers or status

要修改頭部或狀態程式碼，將頭部或狀態引數傳遞給這些函式

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

常用

NotFound:在沒有找到合適的請求路徑時呼叫
ServerError:當伺服器內部出現問題時呼叫。如果在使用者程式碼中出現異常，通常會出現這種情況。

中介軟體和監聽

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

如果想在伺服器啟動或關閉時執行啟動/分解程式碼，可以使用以下偵聽器:

before_server_start
after_server_start
before_server_stop
after_server_stop

這些監聽器在函式中實現為修飾符，它們接受應用程式物件和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,I totally found the page: {}".format(request.url))
複製程式碼

靜態檔案

Static files

靜態檔案可以新增字首

bp.static('/folder/to/serve','/web/path')
複製程式碼

Start and stop

藍圖可以在伺服器的啟動和停止過程中執行函式。如果在多處理器模式下執行(超過1個worker)，這些都是在workers 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()
複製程式碼

用例：API版本控制

Use-case: API versioning

藍圖對於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',debug=True)
複製程式碼

用url_for構建url

如果希望在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是什麼? Keep Alive Timeout value的作用是什麼呢?

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

cookie是儲存在使用者瀏覽器內的資料塊。Sanic既可以讀寫cookie，也可以儲存為鍵-值對。

Warning

cookie可以由客戶機自由修改。因此，您不能僅在cookie中儲存諸如登入資訊這樣的資料，因為客戶機可以隨意更改這些資料。為了確保儲存在cookie中的資料不會被客戶偽造或篡改， use something like itsdangerous to cryptographically sign the data.

讀取cookies

使用者的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))

複製程式碼

寫入cookies

返回響應時，可以在響應物件上設定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
複製程式碼

刪除cookies

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可以設定為字典值，並具有以下引數:

expires (datetime): cookie在客戶機瀏覽器上過期的時間。
path(string): 此cookie應用的url的子集。預設為/。
comment(string): 註釋(元資料)。
domain(string): 指定cookie有效的域。顯式指定的域必須總是以一個點開始。
max-age(number): cookie應該存活的秒數。
secure (boolean): 指定cookie是否只通過HTTPS傳送。
httponly (boolean): 指定Javascript是否不能讀取cookie。

session

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)
複製程式碼

Handler Decorators

由於Sanic處理程式是簡單的Python函式，您可以用類似於Flask的方式向它們應用decorator。一個典型的用例是，當您需要一些程式碼在處理程式的程式碼執行之前執行。

Authorization 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,**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'})
複製程式碼

Streaming

流媒體

Request Streaming

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)
複製程式碼

Response Streaming

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引數

如果您需要任何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,name):
    return text('Hello I have a decorator')

app.add_route(ViewWithDecorator.as_view(),'/url')
複製程式碼

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: 指定請求的最大大小的整數，以位元組為單位。

Example

如果處理函式不返回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',protocol=CustomHttpProtocol)
複製程式碼

SSL Example

可以傳入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",ssl=ssl)
複製程式碼

日誌

Logging

Sanic允許您根據python3 logging API對請求進行不同型別的日誌記錄(訪問日誌、錯誤日誌)。如果您想建立一個新的配置，您應該對python3 logging有一些基本的瞭解。

Quick Start

使用預設設定的一個簡單示例如下:

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_config：

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)
複製程式碼

Configuration

預設情況下，log_config引數設定為使用sanic.log.LOGGING_CONFIG_DEFAULTS字典配置。

在sanic中使用了三個日誌記錄器loggers，如果您想建立自己的日誌配置，則必須定義:

root: 用於記錄內部訊息。

sanic.error: 用於記錄錯誤日誌。

sanic.access: 用於記錄訪問日誌。

Log format

除了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(default True) 一個布林值，它決定原始請求是否由函式返回。如果設定為True，返回值是(request,response)的一個元組，如果False僅返回響應。

server_kwargs (default {}) 在測試請求執行之前傳遞給app.run的附加引數。

debug(default 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',response = app.test_client.post('/',data=json.dumps(data))
    assert request.json.get('key1') == 'value1'
複製程式碼

關於aiohttp的可用引數的更多資訊可以在ClientSession的檔案中找到。

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}
複製程式碼

部署

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',workers=4)
複製程式碼

通過Gunicorn執行

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擴充套件列表。

Sanic-Plugins-Framework: 方便建立和使用Sanic外掛的庫。
Sessions: 對sessions的支援。允許使用redis、memcache或記憶體儲存。
CORS: A port of flask-cors.
Compress: 允許您輕鬆地壓縮Sanic響應。Flask-Compress的一個埠。
Jinja2: 支持Jinja2模板。
JWT: JSON Web令牌(JWT)的身份驗證擴充套件。
OpenAPI/Swagger: OpenAPI支援，外加Swagger UI。
Pagination: 簡單的分頁的支援。
Motor: Simple motor wrapper。
Sanic CRUD: 與peewee模型的CRUD REST API生成。
UserAgent: Add user_agent to request。
Limiter: sanic的速率限制。
Sanic EnvConfig: 將環境變數引入到Sanic配置中。
Babel: 在Babel庫的幫助下向Sanic應用程式新增i18n/l10n支援。
Dispatch: 在werkzeug中由DispatcherMiddleware激發的排程程式。可以充當sanicto - wsgi介面卡。
Sanic-OAuth: OAuth庫，用於連線和建立您自己的令牌提供者。
Sanic-nginx-docker-example: 使用Sanic構建簡單易用的骨架專案需要使用nginx，由docker-compose編排。
sanic-graphql: Graphico與Sanic整合。
sanic-prometheus: Sanic的Prometheus標準。
Sanic-RestPlus: A port of Flask-RestPlus for Sanic. Full-featured REST API with SwaggerUI generation。
sanic-transmute: 從python函式和類中生成api的Sanic擴充套件，並自動生成Swagger UI檔案。
pytest-sanic: Sanic的一個pytest外掛。它幫助您非同步地測試程式碼。
jinja2-sanic: Sanic的jinja2模板渲染器。

API Reference

sanic-官方檔案

sanic-githup

sanic非同步框架之中文檔案

快速開始

安裝

虛擬環境

安裝Sanic

開始

路由

請求引數

請求型別

增加路由

url_for

網路套接字路由

請求

常用型別

使用get和getlist訪問資料

響應

text

HTML

JSON

File

Streaming

File Streaming

Redirect

Raw

Modify headers or status

靜態檔案

模版

異常

丟擲異常

處理異常

有用的異常

中介軟體和監聽

中介軟體

修改請求或響應

監聽者

藍圖

註冊藍圖

使用藍圖

網路套接字路由

中介軟體

異常

靜態檔案

Start and stop

用例：API版本控制

用url_for構建url

配置

基本配置

載入配置

從環境變數

從物件

從檔案

內建配置值

不同的超時變數

Keep-Alive是什麼? Keep Alive Timeout value的作用是什麼呢?

Cookie

讀取cookies

寫入cookies

刪除cookies

session

Handler Decorators

Authorization Decorator

Streaming

Request Streaming

Response Streaming

基於類的檢視

定義檢視

URL引數

裝飾器

URL構建

使用組合檢視

自定義協議

Example

SSL Example

日誌

Quick Start

Configuration

Log format

測試

pytest-sanic

部署

使用`get`和`getlist`訪問資料