摘要：實例實例測試結(jié)果增加路由實例測試結(jié)果提供了一個方法，根據(jù)處理程序方法名生成。異常拋出異常要拋出異常，只需從異常模塊中提出相應(yīng)的異常。

typora-copy-images-to: ipic

[TOC]

在安裝Sanic之前，讓我們一起來看看Python在支持異步的過程中，都經(jīng)歷了哪些比較重大的更新。

首先是Python3.4版本引入了asyncio，這讓Python有了支持異步IO的標(biāo)準(zhǔn)庫，而后3.5版本又提供了兩個新的關(guān)鍵字async/await，目的是為了更好地標(biāo)識異步IO，讓異步編程看起來更加友好，最后3.6版本更進一步，推出了穩(wěn)定版的asyncio，從這一系列的更新可以看出，Python社區(qū)正邁著堅定且穩(wěn)重的步伐向異步編程靠近。

Sanic是一個支持 async/await 語法的異步無阻塞框架，這意味著我們可以依靠其處理異步請求的新特性來提升服務(wù)性能，如果你有Flask框架的使用經(jīng)驗，那么你可以迅速地使用Sanic來構(gòu)建出心中想要的應(yīng)用，并且性能會提升不少，我將同一服務(wù)分別用Flask和Sanic編寫，再將壓測的結(jié)果進行對比，發(fā)現(xiàn)Sanic編寫的服務(wù)大概是Falsk的1.5倍。

僅僅是Sanic的異步特性就讓它的速度得到這么大的提升么？是的，但這個答案并不標(biāo)準(zhǔn)，更為關(guān)鍵的是Sanic使用了uvloop作為asyncio的事件循環(huán)，uvloop由Cython編寫，它的出現(xiàn)讓asyncio更快，快到什么程度？這篇文章中有介紹，其中提出速度至少比 nodejs、gevent 和其他Python異步框架要快兩倍，并且性能接近于用Go編寫的程序，順便一提，Sanic的作者就是受這篇文章影響，這才有了Sanic。

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

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

程序世界一部分是對應(yīng)著現(xiàn)實的，在生活中，我們會在不同的環(huán)境完成不同的任務(wù)，比如在廚房做飯、臥室休息，分工極其明確。

其實用Python編寫應(yīng)用服務(wù)也是如此，它們同樣希望應(yīng)用服務(wù)與開發(fā)環(huán)境是一對一的關(guān)系，這樣做的好處在于，每個獨立的環(huán)境都可以簡潔高效地管理自身對應(yīng)服務(wù)所依賴的第三方庫，如若不然，各個服務(wù)都安排在同一環(huán)境，這樣不僅會造成管理上的麻煩，還會使第三方庫之間產(chǎn)生沖突。

通過上面的敘述，我們是不是可以得出這樣一個核心觀點：應(yīng)該在不同的環(huán)境下做不同的事 ，以此類推，寫項目的時候，我們也需要為每個不同的項目構(gòu)建一個無干擾的的環(huán)境，發(fā)散思維，總結(jié)一下：

不同的項目，需要為其構(gòu)建不同的虛擬環(huán)境，以免互相干擾

構(gòu)建虛擬環(huán)境的工具很多，如下：

virtualenv

pyenv

anaconda

venv

…...

以上三個工具都可以快速地幫助我們構(gòu)建當(dāng)前需要的Python環(huán)境，如果你之前沒有使用過，可直接點開鏈接進行下載，如果你正在使用其它的環(huán)境管理工具，也不要緊，因為不論你使用哪一種方式，我們最終目的都是針對一個新項目構(gòu)建一個新的環(huán)境。

安裝配置好之后，簡單看看官方提供的使用方法，就可以開始了，比如我本機使用的是venv(python3.5以后官方推薦使用這個venv來管理虛擬環(huán)境)，安裝完成后可以很方便地創(chuàng)建一個虛擬環(huán)境，比如這里使用Python3.6來作為本書項目的默認環(huán)境：

cd ~/
# 新建一個python3.6環(huán)境
python3 -m venv pyenv
# 安裝好之后 輸入下面命令進入名為python36的環(huán)境
cd pyenv/
source bin/activate
# 查看版本
python -V

若安裝速度比較慢，可以考慮換國內(nèi)源，比如 國內(nèi)鏡像 ，至于為什么選擇python3.6作為默認環(huán)境，一是因為Sanic只支持Python3.5+，二則是我們構(gòu)建的項目最終是要在生產(chǎn)環(huán)境下運行的，所以建議最好安裝Python3.6下穩(wěn)定版本的asyncio。

Python安裝第三方模塊都是利用pip工具進行安裝，這里也不例外，首先進入上一步我們新建的 python3.6 虛擬環(huán)境，然后安裝：

# 安裝Sanic，請先使用 source activate python36 進入虛擬環(huán)境
pip install sanic
# 如果不想使用uvloop和ujson 可以這樣安裝
SANIC_NO_UVLOOP=true SANIC_NO_UJSON=true pip install sanic

通過上面的命令，你就可以在 python3.6 虛擬環(huán)境中安裝Sanic以及其依賴的第三方庫了，若想查看Sanic是否已經(jīng)正確安裝，可以進入終端下對應(yīng)的虛擬環(huán)境，啟動Python解釋器，導(dǎo)入Sanic庫：

python
>>> 
>>> import sanic

如果沒有出現(xiàn)錯誤，就說明你已經(jīng)正確地安裝了Sanic，請繼續(xù)閱讀下一節(jié)，了解下如何利用Sanic來構(gòu)建一個Web項目吧。

我們將正式使用Sanic來構(gòu)建一個web項目，讓我們踏出第一步，利用Sanic來編寫一個返回Hello World!字符串的服務(wù)程序。

新建一個文件夾sanicweb：

$ mkdir sanicweb
$ cd sanicweb/
$ pwd
/Users/junxi/pyenv/sanicweb

創(chuàng)建一個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的一些基本用法，如路由的構(gòu)建、接受請求數(shù)據(jù)以及返回響應(yīng)的內(nèi)容等。

路由允許用戶為不同的URL端點指定處理程序函數(shù)。

實例:

from sanic.response import json
@app.route("/")
async def index(request):
    return json({ "hello": "world" })

url http://server.url/ 被訪問(服務(wù)器的基本url)，最終/被路由器匹配到處理程序函數(shù)，測試，然后返回一個JSON對象。

必須使用async def語法來定義Sanic處理函數(shù)，因為它們是異步函數(shù)。

要指定一個參數(shù)，可以用像這樣的角引號包圍它。請求參數(shù)將作為關(guān)鍵字參數(shù)傳遞給路線處理程序函數(shù)。

實例：

@app.router("/tag/")
async def tag_handler(request, tag):
    return text("Tag - {}".format(tag))

重啟服務(wù)，輸入地址http://127.0.0.1:9000/tag/python進行訪問

$ curl -X GET http://127.0.0.1:9000/tag/python
Tag - python

為參數(shù)指定類型，在參數(shù)名后面添加（：類型）。如果參數(shù)不匹配指定的類型，Sanic將拋出一個不存在的異常，導(dǎo)致一個404頁面

@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))

測試結(jié)果如下：

$ 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

路由裝飾器接受一個可選的參數(shù)，方法，它允許處理程序函數(shù)與列表中的任何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))

測試結(jié)果：

$ 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/")
app.add_route(personal_handler2, "/personal2/", methods=["GET"])

測試結(jié)果：

$ 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方法，根據(jù)處理程序方法名生成url。避免硬編碼url路徑到您的應(yīng)用程序

實例：

@app.router("/")
async def index(request):
    url = app.url_for("post_handler", post_id=5)
    return redirect(url)


@app.route("posts/")
async def post_handler(request, post_id):
    return text("Post - {}".format(post_id))

給url_for的關(guān)鍵字參數(shù)不是請求參數(shù)，它將包含在URL的查詢字符串中。例如:

url = app.url_for("post_handler", post_id=5, arg_one="one", arg_two="two")
# /posts/5?arg_one=one&arg_two=two

所有有效的參數(shù)必須傳遞給url以便構(gòu)建一個URL。如果沒有提供一個參數(shù)，或者一個參數(shù)與指定的類型不匹配，就會拋出一個URLBuildError

可以將多值參數(shù)傳遞給url

url = app.url_for("post_handler", post_id=5, arg_one=["one", "two"])
# /posts/5?arg_one=one&arg_one=two

經(jīng)過測試訪問/我們會發(fā)現(xiàn)，url跳轉(zhuǎn)到了/posts/5 ，并打印出來Post - 5 的結(jié)果。

redirect是從sanic.response導(dǎo)入的方法，用于處理url的重定向。

WebSocket routes

websocket可以通過裝飾路由實現(xiàn)

實例：

@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

當(dāng)一個端點收到一個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參數(shù)。對于之前的同一個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） - 發(fā)送原始主體。無論內(nèi)容類型如何，該屬性都允許檢索請求的原始數(shù)據(jù)。

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） - 包含請求標(biāo)頭的不區(qū)分大小寫的字典。

ip （str） - 請求者的IP地址。

port （str） - 請求者的端口地址。

socket （tuple） - 請求者的（IP，端口）。

app - 對處理此請求的Sanic應(yīng)用程序?qū)ο蟮囊谩．?dāng)模塊內(nèi)部的藍圖或其他處理程序無法訪問全局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：與請求關(guān)聯(lián)的URL方案：http或https

host：與請求關(guān)聯(lián)的主機： localhost:8080

path：請求的路徑： /posts/1/

query_string：請求的查詢字符串：foo=bar或一個空白字符串""

uri_template：匹配路由處理程序的模板： /posts//

token：授權(quán)標(biāo)頭(Authorization)的值： Basic YWRtaW46YWRtaW4=

返回字典的請求屬性實際上會返回一個dict被調(diào)用的子類 RequestParameters。使用這個對象的關(guān)鍵區(qū)別在于get和getlist方法之間的區(qū)別。

get(key, default=None)按照正常操作，除了當(dāng)給定鍵的值是列表時，只返回第一個項目。

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("
Hello world!
")

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")

沒有進行編碼的響應(yīng)

from sanic import response

@app.route("/raw")
def handle_request(request):
    return response.raw("raw data")

要修改頭部或狀態(tài)代碼，將頭部或狀態(tài)參數(shù)傳遞給這些函數(shù)

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

靜態(tài)文件和目錄，比如一個圖像文件，在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構(gòu)建一個靜態(tài)文件的URL

html templates編寫

編寫web服務(wù)，自然會涉及到html，sanic自帶有html函數(shù)，但這并不能滿足有些需求，故引入jinja2迫在眉睫。使用方法也很簡單：

# novels_blueprint.py片段
from sanic import Blueprint
from jinja2 import Environment, PackageLoader, select_autoescape

# 初始化blueprint并定義靜態(tài)文件夾路徑
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")

這樣，就實現(xiàn)了jinja2 模版的引入。

Exceptions

要拋出異常，只需從sanic異常模塊中提出相應(yīng)的異常。

from sanic.exceptions import ServerError
@app.route("/killme")
def i_am_ready_to_die(request):
    raise ServerError("Something bad happened", status_code=500)

也可以自定義狀態(tài)碼

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

裝飾器一個異常列表作為參數(shù)來處理。你可以通過SanicException來捕獲它們！裝飾異常處理函數(shù)必須將請求和異常對象作為參數(shù)。

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:在沒有找到合適的請求路徑時調(diào)用

ServerError:當(dāng)服務(wù)器內(nèi)部出現(xiàn)問題時調(diào)用。如果在用戶代碼中出現(xiàn)異常，通常會出現(xiàn)這種情況。

Middleware And Listeners

Middleware

有兩種類型的中間件: 請求和響應(yīng)。兩者都是使用@app聲明的。中間件裝飾器，裝飾器的參數(shù)是一個代表其類型的字符串:“請求”或“響應(yīng)”。響應(yīng)中間件接收請求和響應(yīng)作為參數(shù)。

最簡單的中間件根本不修改請求或響應(yīng)

@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")

中間件可以修改它所提供的請求或響應(yīng)參數(shù)，只要它不返回它

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)

上述代碼將按順序應(yīng)用這兩個中間件。首先，中間件custombanner將把HTTP響應(yīng)頭服務(wù)器更改為假服務(wù)器，而第二個中間件防止XSS將添加HTTP頭來防止跨站點腳本攻擊(XSS)攻擊。這兩個函數(shù)是在用戶函數(shù)返回響應(yīng)之后調(diào)用的。

Listeners

如果想在服務(wù)器啟動或關(guān)閉時執(zhí)行啟動/分解代碼，可以使用以下偵聽器:

before_server_start

after_server_start

before_server_stop

after_server_stop

這些監(jiān)聽器在函數(shù)中實現(xiàn)為修飾符，它們接受應(yīng)用程序?qū)ο蠛蚢syncio循環(huán)

@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()

如果你想在循環(huán)開始后運行一個后臺任務(wù)，那么Sanic就提供了addtask方法來輕松地完成這一任務(wù)。

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

藍圖是可以用于應(yīng)用程序中的子路由的對象。除了向應(yīng)用程序?qū)嵗砑勇酚桑{圖還定義了類似的添加路由的方法，然后以靈活的可插入的方式在應(yīng)用程序中注冊。

simple Blueprint

假設(shè)將該文件保存為myblueprint。py，稍后可以導(dǎo)入到您的主應(yīng)用程序中。

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

藍圖必須在應(yīng)用程序中注冊

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

使用藍圖還可以在全局內(nèi)注冊中間件。

@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

靜態(tài)文件可以添加前綴

bp.static("/folder/to/serve", "/web/path")

藍圖可以在服務(wù)器的啟動和停止過程中運行函數(shù)。如果在多處理器模式下運行(超過1個worker)，這些都是在workers fork之后觸發(fā)的。

before_server_start:在服務(wù)器開始接受連接之前執(zhí)行

after_server_start:在服務(wù)器開始接受連接后執(zhí)行

before_server_stop:在服務(wù)器停止接受連接之前執(zhí)行

after_server_stop:在服務(wù)器停止后執(zhí)行，所有請求都完成了

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/，另一個指向/v2/。

當(dāng)一個藍圖被初始化時，它可以選擇一個可選的url_prefix參數(shù)，這個參數(shù)將被預(yù)先定義到藍圖中定義的所有路由。該特性可用于實現(xiàn)我們的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")

當(dāng)我們在應(yīng)用程序上注冊我們的藍圖時，路徑/v1和/v2將指向單個的藍圖，它允許為每個API版本創(chuàng)建子站點。

# 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內(nèi)部路由生成一個URL，記住，端點名稱采用格式.

@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/")
async def post_handler(request, post_id):
    return text("Post {} in Blueprint V1".format(post_id))

Note: 當(dāng)app和blueprint不在同一個模塊內(nèi)記得加上request

例如：url = request.app.url_for("v1.post_handler", post_id=5) # --> "/v1/post/5"

Configuration

任何一個相當(dāng)復(fù)雜的應(yīng)用程序都需要配置，而不是在實際代碼中進行。對于不同的環(huán)境或安裝，設(shè)置可能是不同的。

Sanic在應(yīng)用程序?qū)ο蟮呐渲脤傩灾斜３峙渲谩Ｅ渲脤ο髢H僅是一個可以使用點符號或字典來修改的對象。

app = Sanic("myapp")
app.config.DB_NAME = "appdb"
app.config.DB_USER = "appuser"

因為配置對象實際上是一個字典，所以可以使用它的update方法來一次設(shè)置幾個值:

db_settings = {
    "DB_HOST": "localhost",
    "DB_NAME": "appdb",
    "DB_USER": "appuser"
}
app.config.update(db_settings)

一般來說，該約定只具有大寫的配置參數(shù)。下面描述的加載配置的方法只會查找這些大寫的參數(shù)。

如何加載配置有幾種方法。

使用SANIC_前綴定義的任何變量都將應(yīng)用于sanic config。例如，設(shè)置SANIC_REQUEST_TIMEOUT將由應(yīng)用程序自動加載并輸入REQUEST_TIMEOUT配置變量。你可以通過一個不同的前綴到Sanic:

app = Sanic(load_env="MYAPP_")

然后，上面的變量將是MYAPP_REQUEST_TIMEOUT。如果您想要禁用環(huán)境變量的加載，您可以將其設(shè)置為False:

app = Sanic(load_env=False)

如果有很多配置值，而且它們有合理的默認值，那么將它們放到一個模塊中可能會有幫助:

import myapp.default_settings

app = Sanic("myapp")
app.config.from_object(myapp.default_settings)

您也可以使用類或任何其他對象。

通常，您需要從一個不屬于分布式應(yīng)用程序的文件中加載配置。可以使用from_pyfile(/path/to/config_file)從文件加載配置。但是，這需要程序知道配置文件的路徑。因此，您可以在一個環(huán)境變量中指定配置文件的位置，并告訴Sanic使用它來查找配置文件:

app = Sanic("myapp")
app.config.from_envvar("MYAPP_SETTINGS")

然后，您可以使用MYAPP_SETTINGS環(huán)境變量集運行您的應(yīng)用程序:

$ MYAPP_SETTINGS=/path/to/config_file python3 myapp.py
INFO: Goin" Fast @ http://0.0.0.0:8000

配置文件是常規(guī)的Python文件，這些文件是為了加載它們而執(zhí)行的。這允許您使用任意邏輯來構(gòu)造正確的配置。在配置中只添加了大寫的變量。最常見的配置包括簡單的鍵值對：

# config_file
DB_HOST = "localhost"
DB_NAME = "appdb"
DB_USER = "appuser"

在這個框中，只有幾個預(yù)定義值，在創(chuàng)建應(yīng)用程序時可以重寫。

| 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后端服務(wù)器時的時間間隔，以及接收整個HTTP請求的瞬間。如果時間超過了REQUEST_TIMEOUT值(以秒為單位)，那么這將被視為客戶端錯誤，因此Sanic生成一個HTTP 408響應(yīng)并將其發(fā)送給客戶端。如果您的客戶經(jīng)常通過非常大的請求負載或者非常緩慢地上傳請求，請調(diào)整這個值。

響應(yīng)超時度量在Sanic服務(wù)器將HTTP請求傳遞到Sanic應(yīng)用程序的時間之間的時間，以及發(fā)送到客戶機的HTTP響應(yīng)的時間。如果時間超過了RESPONSE_TIMEOUT值(以秒為單位)，這被認為是服務(wù)器錯誤，因此Sanic生成一個HTTP 503響應(yīng)并將其設(shè)置為客戶機。如果應(yīng)用程序很可能長時間運行，延遲響應(yīng)的生成，則將此值調(diào)整得更高。

Keep-Alive是HTTP 1.1中的一個HTTP特性。發(fā)送HTTP請求時，客戶端(通常是web瀏覽器應(yīng)用程序)可以設(shè)置一個keepalive消息頭，以指示HTTP服務(wù)器(Sanic)在發(fā)送響應(yīng)之后不關(guān)閉TCP連接。這允許客戶端重用現(xiàn)有的TCP連接來發(fā)送后續(xù)的HTTP請求，并確保客戶機和服務(wù)器的網(wǎng)絡(luò)流量更高效。

在Sanic中，KEEP_ALIVE配置變量默認設(shè)置為True。如果您在應(yīng)用程序中不需要此功能，則將其設(shè)置為False，以便在發(fā)送響應(yīng)后立即關(guān)閉所有客戶端連接，而不考慮請求上的keepalive消息頭。

服務(wù)器保持TCP連接打開的時間量由服務(wù)器本身決定。在Sanic中，該值使用KEEP_ALIVE_TIMEOUT值進行配置。默認情況下，它設(shè)置為5秒，這是與Apache HTTP服務(wù)器相同的默認設(shè)置，在允許客戶端發(fā)送新請求的足夠時間和不同時打開太多連接之間保持良好的平衡。不要超過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是保存在用戶瀏覽器內(nèi)的數(shù)據(jù)塊。Sanic既可以讀寫cookie，也可以存儲為鍵-值對。

Warning

cookie可以由客戶機自由修改。因此，您不能僅在cookie中存儲諸如登錄信息這樣的數(shù)據(jù)，因為客戶機可以隨意更改這些數(shù)據(jù)。為了確保存儲在cookie中的數(shù)據(jù)不會被客戶偽造或篡改， 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))

返回響應(yīng)時，可以在響應(yīng)對象上設(shè)置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

響應(yīng)cookie可以設(shè)置為字典值，并具有以下參數(shù):

expires (datetime): cookie在客戶機瀏覽器上過期的時間。

path(string): 此cookie應(yīng)用的url的子集。默認為/。

comment(string): 注釋(元數(shù)據(jù))。

domain(string): 指定cookie有效的域。顯式指定的域必須總是以一個點開始。

max-age(number): cookie應(yīng)該存活的秒數(shù)。

secure (boolean): 指定cookie是否只通過HTTPS發(fā)送。

httponly (boolean): 指定Javascript是否不能讀取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函數(shù)，您可以用類似于Flask的方式向它們應(yīng)用decorator。一個典型的用例是，當(dāng)您需要一些代碼在處理程序的代碼執(zhí)行之前運行。

假設(shè)您想要檢查用戶是否被授權(quán)訪問某個特定的端點。您可以創(chuàng)建包裝處理函數(shù)的decorator，檢查客戶端是否被授權(quán)訪問某個資源，并發(fā)送適當(dāng)?shù)捻憫?yīng)。

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允許您通過流獲取請求數(shù)據(jù)，如下所示。當(dāng)請求結(jié)束時，request.stream.get()返回None。只有post, put和patch decorator 有流參數(shù)。

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 方法將內(nèi)容流到客戶機。該方法接受一個coroutine回調(diào)，該回調(diào)將傳遞給寫入的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")

在您希望將內(nèi)容流到來自外部服務(wù)(如數(shù)據(jù)庫)的客戶端時，這很有用。例如，您可以使用asyncpg提供的異步游標(biāo)將數(shù)據(jù)庫記錄流到客戶端:

@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)

基于類的視圖只是實現(xiàn)對請求的響應(yīng)行為的類。它們提供了一種方法，將不同的HTTP請求類型劃分到同一端點。與其定義和修飾三個不同的處理函數(shù)(每個端點支持的請求類型)，端點可以分配一個基于類的視圖。

基于類的視圖應(yīng)該子類化HTTPMethodView。然后，您可以為希望支持的每個HTTP請求類型實現(xiàn)類方法。如果接收到的請求沒有定義的方法，則會生成一個405: Method not allowed 的響應(yīng)。

要在端點上注冊基于類的視圖，將使用app.add_route方法。第一個參數(shù)應(yīng)該是被調(diào)用的方法as_view的定義類，第二個參數(shù)應(yīng)該是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參數(shù)，如路由指南中所討論的，在方法定義中包含它們。

class NameView(HTTPMethodView):

  def get(self, request, name):
    return text("Hello {}".format(name))

app.add_route(NameView.as_view(), "/")

如果您想在類中添加任何裝飾器，可以設(shè)置decorator類變量。當(dāng)調(diào)用as_view時，這些將被應(yīng)用到類中。

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構(gòu)建一個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將處理程序函數(shù)移到視圖類之外。

每個支持的HTTP方法的處理函數(shù)都在源代碼的其他地方定義，然后使用CompositionView.add方法添加到視圖中。第一個參數(shù)是要處理的HTTP方法的列表(例如，["GET"， "POST"])，第二個參數(shù)是處理函數(shù)。下面的示例顯示了使用外部處理程序函數(shù)和內(nèi)聯(lián)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: 當(dāng)前您不能使用url_for為CompositionView構(gòu)建一個URL。

注意:這是高級用法，大多數(shù)讀者不需要這樣的功能。

您可以通過指定自定義協(xié)議來更改Sanic協(xié)議的行為，該協(xié)議應(yīng)該是asyncio.protocol的子類。然后，該協(xié)議可以作為sanic.run方法的關(guān)鍵字參數(shù)協(xié)議傳遞。

自定義協(xié)議類的構(gòu)造函數(shù)接收來自Sanic的以下關(guān)鍵字參數(shù)。

loop: 一個異步兼容的事件循環(huán)。

connections: 用于存儲協(xié)議對象的集合。當(dāng)Sanic接收SIGINT或SIGTERM時，它執(zhí)行protocol.close_if_idle關(guān)閉此集合中存儲的所有協(xié)議對象。

signal: 帶有stopped屬性的sanic.server.Signal對象。當(dāng)Sanic收到SIGINT或SIGTERM，signal.stopped分配True。

request_handler: 取一個sanic.request.Request對象和response回調(diào)作為參數(shù)的coroutine。

error_handler: 在拋出異常時調(diào)用的處理程序sanic.exceptions.Handler。

request_timeout: 請求超時前的秒數(shù)。

request_max_size: 指定請求的最大大小的整數(shù)，以字節(jié)為單位。

如果處理函數(shù)不返回HTTPResponse對象，則默認協(xié)議中出現(xiàn)錯誤。

通過重寫write_response協(xié)議方法，如果處理程序返回一個字符串，它將被轉(zhuǎn)換為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允許您根據(jù)python3 logging API對請求進行不同類型的日志記錄(訪問日志、錯誤日志)。如果您想創(chuàng)建一個新的配置，您應(yīng)該對python3 logging有一些基本的了解。

使用默認設(shè)置的一個簡單示例如下:

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應(yīng)用時傳遞log_config：

app = Sanic("test", log_config=LOGGING_CONFIG)

要關(guān)閉日志，只需分配access_log=False:

if __name__ == "__main__":
  app.run(access_log=False)

這將跳過在處理請求時調(diào)用日志功能。你甚至可以做進一步的生產(chǎn)以獲得額外的速度:

if __name__ == "__main__":
  # disable debug messages
  app.run(debug=False, access_log=False)

默認情況下，log_config參數(shù)設(shè)置為使用sanic.log.LOGGING_CONFIG_DEFAULTS字典配置。

在sanic中使用了三個日志記錄器loggers，如果您想創(chuàng)建自己的日志配置，則必須定義:

root: 用于記錄內(nèi)部消息。
sanic.error: 用于記錄錯誤日志。
sanic.access: 用于記錄訪問日志。

除了python提供的默認參數(shù)(asctime、levelname、message)， Sanic還提供了用于訪問日志記錄器logger的其他參數(shù):

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方法，以便與應(yīng)用程序運行。一個簡單的例子(使用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

在內(nèi)部，每次調(diào)用test_client方法時，Sanic應(yīng)用程序運行于127.0.0.1:42101，您的測試請求使用aiohttp執(zhí)行。

test_client方法接受以下參數(shù)和關(guān)鍵字參數(shù):

uri(default "/") 一個表示測試uri的字符串。
gather_request(default  True) 一個布爾值，它決定原始請求是否由函數(shù)返回。如果設(shè)置為True，返回值是(request, response)的一個元組，如果False僅返回響應(yīng)。
server_kwargs (default {}) 在測試請求運行之前傳遞給app.run的附加參數(shù)。
debug(default False)一個布爾值，它決定是否在調(diào)試模式下運行服務(wù)器。

該函數(shù)進一步接受了*request_args和**request_kwargs，它們直接傳遞給aiohttp ClientSession請求。

例如，為了向GET請求提供數(shù)據(jù)，您將執(zhí)行以下操作:

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請求提供數(shù)據(jù):

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"

關(guān)于aiohttp的可用參數(shù)的更多信息可以在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還提供了一些有用的設(shè)備，如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是由內(nèi)置的webserver簡化的。在定義了sanic.Sanic的實例之后。我們可以用以下關(guān)鍵字參數(shù)調(diào)用run方法:

host (default "127.0.0.1"): 地址來托管服務(wù)器。

port (default 8000): 開啟服務(wù)器的端口。

debug (default False): 啟用調(diào)試輸出(減慢服務(wù)器)。

ssl (default None): ssl加密的SSLContext。

sock (default None): 用于服務(wù)器接受連接的套接字。

workers (default 1): 生成的工作進程數(shù)。

loop (default None): 一個asyncio兼容的事件循環(huán)。如果沒有指定，Sanic將創(chuàng)建自己的事件循環(huán)。

protocol (default HttpProtocol): asyncio.protocol的子類。

Workers

默認情況下，Sanic只使用一個CPU核心偵聽主進程。To crank up the juice，只需在run參數(shù)中指定workers的數(shù)量。

app.run(host="0.0.0.0", port=1337, workers=4)

Sanic將會自動啟動多個進程，并在它們之間路由流量。我們建議盡可能多的workers擁有可用的核心。

如果您喜歡使用命令行參數(shù)，則可以通過執(zhí)行模塊來啟動Sanic服務(wù)器。例如，如果您在名為server.py的文件中初始化Sanic作為app，你可以這樣運行服務(wù)器:

python -m sanic server.app --host=0.0.0.0 --port=1337 --workers=4

通過這種運行sanic的方式，無需在Python文件中調(diào)用app.run。如果這樣做，請確保將其包裝起來，以便它只在解釋器直接運行時執(zhí)行。

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=1337, workers=4)

Gunicorn ‘Green Unicorn’ 是UNIX的一個WSGI HTTP服務(wù)器。它是一個由Ruby的Unicorn項目移植的預(yù)fork worker模型。

為了使用Gunicorn運行Sanic應(yīng)用程序，您需要為Gunicornworker-class 參數(shù)使用特殊的sanic.worker.GunicornWorker：

gunicorn myapp:app --bind 0.0.0.0:1337 --worker-class sanic.worker.GunicornWorker

如果您的應(yīng)用程序遭受內(nèi)存泄漏，您可以配置Gunicorn以優(yōu)雅地重新啟動一個worker，因為它處理了給定數(shù)量的請求。這可以方便地幫助限制內(nèi)存泄漏的影響。

有關(guān)更多信息，請參見 Gunicorn 文檔。

Asynchronous support

如果您需要與其他應(yīng)用程序共享sanic進程，特別是loop，這是合適的。但是，請注意，該方法不支持使用多進程，并且不是一般運行該應(yīng)用程序的首選方式。

這里有一個不完整的示例(請參見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

由社區(qū)創(chuàng)建的Sanic擴展列表。

Sanic-Plugins-Framework: 方便創(chuàng)建和使用Sanic插件的庫。

Sessions: 對sessions的支持。允許使用redis、memcache或內(nèi)存存儲。

CORS: A port of flask-cors.

Compress: 允許您輕松地壓縮Sanic響應(yīng)。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: 將環(huán)境變量引入到Sanic配置中。

Babel: 在Babel庫的幫助下向Sanic應(yīng)用程序添加i18n/l10n支持。

Dispatch: 在werkzeug中由DispatcherMiddleware激發(fā)的調(diào)度程序。可以充當(dāng)sanicto - wsgi適配器。

Sanic-OAuth: OAuth庫，用于連接和創(chuàng)建您自己的令牌提供者。

Sanic-nginx-docker-example: 使用Sanic構(gòu)建簡單易用的骨架項目需要使用nginx，由docker-compose編排。

sanic-graphql: Graphico與Sanic集成。

sanic-prometheus: Sanic的Prometheus標(biāo)準(zhǔn)。

Sanic-RestPlus: A port of Flask-RestPlus for Sanic. Full-featured REST API with SwaggerUI generation。

sanic-transmute: 從python函數(shù)和類中生成api的Sanic擴展，并自動生成Swagger UI文檔。

pytest-sanic: Sanic的一個pytest插件。它幫助您異步地測試代碼。

jinja2-sanic: Sanic的jinja2模板渲染器。

sanic-官方文檔

sanic-githup