Sanic框架

Sanic框架

1. 入門

Sanic 是一款相似Flask的Web服務器，它運行在Python 3.5+上。

除了與Flask功能相似以外，它還支持異步請求處理，這意味着你可使用Python3.5 中新的異步/等待語法，使你的程序運行更加快速。

1.1 簡單起步

from sanic import Sanic
from sanic.response import json

app = Sanic()

@app.route("/")
async def test(request):
    return json({"hello": "world"})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

1. 保存到main.py文件，運行文件python3 main.py；
2. 打開URLhttp://0.0.0.0:8000，能夠看到網頁顯示 Hello World信息。

2. 路由

路由容許用戶爲不一樣的URL地址指定處理的函數。

一個基本的路由就像下面的例子，而app就是Sanic類的一個實例。

from sanic.response import json

@app.route("/")
async def test(request):
    return json({ "hello": "world" })

當地址http://server.url/被訪問時（服務的基礎地址），根地址/就會被路由匹配一個定義了返回JSON對象的test函數。

必須使用async def語法定義函數，來保證其能夠進行異步處理。

2.1 請求參數

Sanic的基礎路由支持請求參數的操做。

若是須要指定參數，請使用尖括號<PARAM>將指定參數括起來。請求參數將做爲路由函數的關鍵字參數。

from sanic.response import text

@app.route('/tag/<tag>')
async def tag_handler(request, tag):
    return text('Tag - {}'.format(tag))

若是須要指定添加的參數的類型，則要在參數名字後面添加:type指定參數類型。若是參數與指定的參數類型不匹配，則Sanic會拋出NotFound的異常，從而致使頁面出現404: Page not found的錯誤。

from sanic.response import text

@app.route('/number/<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))

2.2 HTTP 請求類型

默認狀況下，一個路由會定義一個僅僅適用於URL的GET請求。然而，@app.route裝飾器接受一個可選的參數methods，它容許定義的函數使用列表中任何一個的HTTP方法。

from sanic.response import text

@app.route('/post', methods=['POST'])
async def post_handler(request):
    return text('POST request - {}'.format(request.json))

@app.route('/get', methods=['GET'])
async def get_handler(request):
    return text('GET request - {}'.format(request.args))

這裏還有一個可選的host參數（列表或是字符串）。它限制了給主機的路由。若是存在一個沒有主機的路由，它將是一個默認值。

@app.route('/get', methods=['GET'], host='example.com')
async def get_handler(request):
    return text('GET request - {}'.format(request.args))

# if the host header doesn't match example.com, this route will be used
@app.route('/get', methods=['GET'])
async def get_handler(request):
    return text('GET request in default - {}'.format(request.args))

這裏還有一種快速使用裝飾器的方法：

from sanic.response import text

@app.post('/post')
async def post_handler(request):
    return text('POST request - {}'.format(request.json))

@app.get('/get')
async def get_handler(request):
    return text('GET request - {}'.format(request.args))

2.3 add_route方法

就像上文提到的，路由一般使用@app.route裝飾器進行添加的。可是，這個裝飾器只是app.add_route方法的一個封裝。它看起來像下面這樣：

from sanic.response import text

# Define the handler functions
async def handler1(request):
    return text('OK')

async def handler2(request, name):
    return text('Folder - {}'.format(name))

async def person_handler2(request, name):
    return text('Person - {}'.format(name))

# Add each handler function as a route
app.add_route(handler1, '/test')
app.add_route(handler2, '/folder/<name>')
app.add_route(person_handler2, '/person/<name:[A-z]>', methods=['GET'])

2.4 利用url_for生成URL

Sanic提供了一個根據處理函數名字生成URL的方法url_for。在應用中，它使用處理程序的名字來有效避免使用實際的網絡路徑。

@app.route('/')
async def index(request):
    # generate a URL for the endpoint `post_handler`
    url = app.url_for('post_handler', post_id=5)
    # the URL is `/posts/5`, redirect to it
    return redirect(url)


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

使用url_for須要注意的是:

• 傳遞給url_for的關鍵字若是不是請求參數，將包含在URL 的查詢字符串中。

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

• 能夠傳遞多個參數給url_for函數。

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

• 還能夠傳遞一些特殊的參數給url_for方法來構造一些特殊的URL，諸如：_anchor,_external,_scheme,_server。

url = app.url_for('post_handler', post_id=5, arg_one='one', _anchor='anchor')
# /posts/5?arg_one=one#anchor

url = app.url_for('post_handler', post_id=5, arg_one='one', _external=True)
# //server/posts/5?arg_one=one
# _external requires passed argument _server or SERVER_NAME in app.config or url will be same as no _external

url = app.url_for('post_handler', post_id=5, arg_one='one', _scheme='http', _external=True)
# http://server/posts/5?arg_one=one
# when specifying _scheme, _external must be True

# you can pass all special arguments one time
url = app.url_for('post_handler', post_id=5, arg_one=['one', 'two'], arg_two=2, _anchor='anchor', _scheme='http', _external=True, _server='another_server:8888')
# http://another_server:8888/posts/5?arg_one=one&arg_one=two&arg_two=2#anchor

• 全部的參數都必須正確地傳遞給url_for方法來構造URL。若是未提供參數或指定參數不匹配，將拋出URLBuildError的錯誤。

2.5 WebSocket 路由

使用@app.websocket裝飾器定義WebSocket協議的路由。

@app.websocket('/feed')
async def feed(request, ws):
    while True:
        data = 'hello!'
        print('Sending: ' + data)
        await ws.send(data)
        data = await ws.recv()
        print('Received: ' + data)

或者，使用app.add_websocket_route方法來代替@app.websocket裝飾器。

async def feed(request, ws):
    pass

app.add_websocket_route(my_websocket_handler, '/feed')

WebSocket路由的處理程序將請求做爲第一個參數傳遞，並將WebSocket協議對象做爲第二個參數傳遞。而協議對象具備send和recv兩個方法來進行數據的傳送和接收。

3. 請求數據

當接收端接收到一個HTTP請求的時候，路由函數就會傳遞一個Request對象。

如下的變量能夠做爲Request對象的屬性進行訪問。

• json(任何類型)-JSON格式的數據

from sanic.response import json

@app.route("/json")
def post_json(request):
    return json({ "received": True, "message": request.json })

• arg(dict類型)-查詢字符串變量。一個查詢的字符串是部分的URL，相似於?key1=value1&key2=value2，若是要解析這個URL，那麼arg字典看起來就像{'key1': ['value1'], 'key2': ['value2']}，這個請求將使用query_string變量來保存未解析的字符串的值。

from sanic.response import json

@app.route("/query_string")
def query_string(request):
    return json({ "parsed": True, "args": request.args, "url": request.url, "query_string": request.query_string })

• raw_args(dict類型)-在許多狀況下，你只須要訪問一段很小的url參數。對於前文的URL的?key1=value1&key2=value2，raw_args字典看起來就像這樣{'key1': 'value1', 'key2': 'value2'}。

• files(File對象的字典)-具備名稱，正文和類型的文件列表。

from sanic.response import json

@app.route("/files")
def post_json(request):
    test_file = request.files.get('test')

    file_parameters = {
        'body': test_file.body,
        'name': test_file.name,
        'type': test_file.type,
    }

    return json({ "received": True, "file_names": request.files.keys(), "test_file_parameters": file_parameters })

• form(dict類型)-發佈表單數據

from sanic.response import json

@app.route("/form")
def post_json(request):
    return json({ "received": True, "form_data": request.form, "test": request.form.get('test') })

• body(字節類型)-發佈正文。這個屬性容許檢索請求的原始數據，而無需理會數據的類型。

from sanic.response import text

@app.route("/users", methods=["POST",])
def create_user(request):
    return text("You are trying to create a user with the following POST: %s" % request.body)

• ip(str類型)-請求者的IP地址

• app-對正在處理此請求的Sanic應用程序對象的引用。在沒法訪問全局app對象或其它處理程序的時候頗有用。

from sanic.response import json
from sanic import Blueprint

bp = Blueprint('my_blueprint')

@bp.route('/')
async def bp_root(request):
    if request.app.config['DEBUG']:
        return json({'status': 'debug'})
    else:
        return json({'status': 'production'})

• url:完整的請求URL，即：http://localhost:8000/posts/1/?foo=bar。

• scheme:與請求相關聯的URL方案，http或是https。

• host:與請求相關聯的主機，localhost:8080。

• path:請求的地址/post/1

• query_string:請求的查詢字符串，foo=bar或是空字符串''。

3.1 使用get和getlist來訪問值

請求的屬性其實是返回一個dict的子類RequestParameters。在使用這個對象的主要區別是get和getlist方法的不一樣。

• get(key, default=None) 當給定key的值是一個列表的時候，只返回第一個項目。

• getlist(key, default=None) 返回整個列表。

from sanic.request import RequestParameters

args = RequestParameters()
args['titles'] = ['Post 1', 'Post 2']

args.get('titles') # => 'Post 1'

args.getlist('titles') # => ['Post 1', 'Post 2']

4. 響應

使用sanic.response模塊中的函數來建立響應。

4.1 純文本

from sanic import response

@app.route('/text')
def handle_request(request):
    return response.text('Hello world!')

4.2 HTML

from sanic import response

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

4.3 JSON

from sanic import response

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

4.4 文件

from sanic import response

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

4.5 Streaming

from sanic import response

@app.route("/streaming")
async def index(request):
    async def streaming_fn(response):
        response.write('foo')
        response.write('bar')
    return response.stream(streaming_fn, content_type='text/plain')

4.6 重定向

from sanic import response

@app.route('/redirect')
def handle_request(request):
    return response.redirect('/json')

4.7 元數據

響應未編碼的文本

from sanic import response

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

4.8 修改標題或狀態

要修改標題或狀態，請將標題或狀態參數傳遞給這些函數：

from sanic import response

@app.route('/json')
def handle_request(request):
    return response.json(
        {'message': 'Hello world!'},
        headers={'X-Served-By': 'sanic'},
        status=200
    )

5. 靜態文件

靜態文件和目錄，例如圖像文件是在Sanic建立app.static時候提供的。這個方法採用一個URL地址和一個文件名。而後經過給定的端點訪問指定的文件。

from sanic import Sanic
app = Sanic(__name__)

# Serves files from the static folder to the URL /static
app.static('/static', './static')

# Serves the file /home/ubuntu/test.png when the URL /the_best.png
# is requested
app.static('/the_best.png', '/home/ubuntu/test.png')

app.run(host="0.0.0.0", port=8000)

目前還沒法利用url_for爲靜態文件建立一個URL地址。

6. 異常

異常能夠從請求處理程序中拋出，並由Sanic自動處理。異常將消息做爲第一個參數，也能夠在HTTP響應中傳回狀態代碼。

6.1 拋出異常

要拋出一個異常，只須要從sanic.exceptions中導入與raise相關的異常。

from sanic.exceptions import ServerError

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

6.2 處理異常

若是須要覆蓋Sanic對異常的默認處理，就須要使用@app.exception裝飾器。裝飾器指望使用一個異常列表來處理參數。你能夠傳遞一個SanicException來捕捉它們。裝飾器異常處理函數必須使用Request和Exception對象來做爲參數。

from sanic.response import text
from sanic.exceptions import NotFound

@app.exception(NotFound)
def ignore_404s(request, exception):
    return text("Yep, I totally found the page: {}".format(request.url))

6.3 有用的異常

一些有用的異常以下：

• NotFound: 找不到合適的路由請求。

• ServerError: 服務器內部出現問題時調用。一般發生在用戶代碼出現錯誤的狀況。

7. 中間件和監聽器

中間件是在向服務器請求以前或以後執行的功能。它們可用來修改用戶自定義處理函數的請求或響應。

另外，Sanic提供程序監聽器來運行應用程序生命週期中各個不一樣點的代碼。

7.1 中間件

這裏有兩種不一樣類型的中間件：請求request和響應response。 都是使用@app.middleware裝飾器進行聲明的，利用'request'或'response'字符串來表示其參數類型。

最簡單的中間件不修改任何的請求或響應：

@app.middleware('request')
async def print_on_request(request):
    print("I print when a request is received by the server")

@app.middleware('response')
async def print_on_response(request, response):
    print("I print when a response is returned by the server")

7.2 修改請求或響應

只要請求或修改不返回任何值，中間件就能夠修改給定的請求或響應。下面的示例便是一個簡單的示範：

app = Sanic(__name__)

@app.middleware('response')
async def custom_banner(request, response):
    response.headers["Server"] = "Fake-Server"

@app.middleware('response')
async def prevent_xss(request, response):
    response.headers["x-xss-protection"] = "1; mode=block"

app.run(host="0.0.0.0", port=8000)

上面的代碼將按順序應用兩個中間件。首先，中間件custom_banner將HTTP響應頭服務器更改成Fake-Server，第二個中間件prevent_xss將添加HTTP頭以防止跨站點腳本（XSS）攻擊。這兩個函數在用戶函數返回響應以後調用。

若是中間件返回一個HTTPResponse對象，這個請求將中止處理並返回響應。若是這個請求在相關用戶路由處理到達以前發生，則不會被調用該處理程序。返回的響應還會阻止進一步的中間件運行。

@app.middleware('request')
async def halt_request(request):
    return text('I halted the request')

@app.middleware('response')
async def halt_response(request, response):
    return text('I halted the response')

7.3 監聽器

若是你想要在服務啓動或關閉時執行啓動/拆卸代碼，可使用如下的監聽器：

• before_server_start

• after_server_start

• before_server_stop

• after_server_stop

這些監聽器在接收app對象和asyncio循環的函數上實現爲裝飾器。以下所示：

@app.listener('before_server_start')
async def setup_db(app, loop):
    app.db = await db_setup()

@app.listener('after_server_start')
async def notify_server_started(app, loop):
    print('Server successfully started!')

@app.listener('before_server_stop')
async def notify_server_stopping(app, loop):
    print('Server shutting down!')

@app.listener('after_server_stop')
async def close_db(app, loop):
    await app.db.close()

若是你想要在循環開始後安排後臺容許任務，則可使用add_task方法輕鬆實現。

async def notify_server_started_after_five_seconds():
    await asyncio.sleep(5)
    print('Server successfully started!')

app.add_task(notify_server_started_after_five_seconds())

8. 藍圖

藍圖是能夠用於應用程序中的子路由對象。藍圖不是嚮應用程序實例添加路由，而是定義了相似添加路由的方法，而後將路由以靈活且可插拔的方式註冊到應用程序中。

藍圖對於較大的應用程序十分有用，你可在邏輯上將應用程序分爲幾個組或責任領域。

8.1 第一個藍圖

下面顯示了一個很是簡單的藍圖，它在你的應用程序的根目錄/下注冊了一個處理函數。

假設你將其保存到了my_blueprint.py文件中，則能夠將其導入到你的主應用程序中。

from sanic.response import json
from sanic import Blueprint

bp = Blueprint('my_blueprint')

@bp.route('/')
async def bp_root(request):
    return json({'my': 'blueprint'})

8.2 註冊藍圖

藍圖必須在應用程序中註冊。

from sanic import Sanic
from my_blueprint import bp

app = Sanic(__name__)
app.blueprint(bp)

app.run(host='0.0.0.0', port=8000, debug=True)

藍圖將添加到應用程序，並註冊由該藍圖定義的任何路由。在此示例中，app.router中的註冊路由將以下所示：

[Route(handler=<function bp_root at 0x7f908382f9d8>, methods=None, pattern=re.compile('^/$'), parameters=[])]

8.3 使用藍圖

藍圖與應用程序有着大體相同的功能。

8.3.1 WebSocket協議路由

可使用@bp.websocket裝飾器或bp.add_websocket_route方法在藍圖上註冊WebSocket處理程序。

8.3.2 中間件

使用藍圖能夠在全局註冊中間件。

@bp.middleware
async def print_on_request(request):
    print("I am a spy")

@bp.middleware('request')
async def halt_request(request):
    return text('I halted the request')

@bp.middleware('response')
async def halt_response(request, response):
    return text('I halted the response')

8.3.3 異常

只利用藍圖來應用全局的異常。

@bp.exception(NotFound)
def ignore_404s(request, exception):
    return text("Yep, I totally found the page: {}".format(request.url))

8.3.4 靜態文件

靜態文件能夠在藍圖定義下提供給全局。

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

8.4 啓動和中止

藍圖能夠在服務啓動或中止以前運行功能。若是多進程模式運行（超過1個進程），這些將在進程fork以後被觸發。

可用的事件是：

• before_server_start: 服務開始接受鏈接以前執行

• after_server_start: 服務開始接受鏈接後執行

• before_server_stop: 服務中止接受鏈接以前執行

• after_server_stop: 服務中止而且全部請求完成後執行

bp = Blueprint('my_blueprint')

@bp.listener('before_server_start')
async def setup_connection(app, loop):
    global database
    database = mysql.connect(host='127.0.0.1'...)

@bp.listener('after_server_stop')
async def close_connection(app, loop):
    await database.close()

8.5 用例：API版本控制

Blueprints對於API版本控制很是有用，其中一個藍圖可能指向/v1/<routes>，另外一個指向/v2/<routes>。

當藍圖被初始化時，它可使用一個可選的url_prefix參數，這個參數將被添加到藍圖上定義的全部路由上。此功能可用於實現API版本控制。

# blueprints.py
from sanic.response import text
from sanic import Blueprint

blueprint_v1 = Blueprint('v1', url_prefix='/v1')
blueprint_v2 = Blueprint('v2', url_prefix='/v2')

@blueprint_v1.route('/')
async def api_v1_root(request):
    return text('Welcome to version 1 of our documentation')

@blueprint_v2.route('/')
async def api_v2_root(request):
    return text('Welcome to version 2 of our documentation')

當在應用程序上註冊藍圖時，路由/v1和/v2如今將指向單個藍圖，這容許爲每一個API版本建立子站點。

# main.py
from sanic import Sanic
from blueprints import blueprint_v1, blueprint_v2

app = Sanic(__name__)
app.blueprint(blueprint_v1, url_prefix='/v1')
app.blueprint(blueprint_v2, url_prefix='/v2')

app.run(host='0.0.0.0', port=8000, debug=True)

8.6 利用url_for生成URL

若是但願爲藍圖中的路由生成URL，請記住端點名稱採用的格式<blueprint_name>，<handler_name>。例如：

@blueprint_v1.route('/')
async def root(request):
    url = app.url_for('v1.post_handler', post_id=5) # --> '/v1/post/5'
    return redirect(url)

@blueprint_v1.route('/post/<post_id>')
async def post_handler(request, post_id):
    return text('Post {} in Blueprint V1'.format(post_id))

9. 配置

任何複雜的應用程序都須要合理的配置。不一樣的環境或安裝的設置可能不一樣。

9.1 基本配置

Sanic將配置保存在config應用程序對象的屬性中。配置的是一個可使用點運算進行修改或是相似字典類型的對象。

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

因爲配置的對象其實是一個字典，你可使用update方法來一次性設置幾個值。

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

通常慣例是隻有UPPERCASE配置參數。下面描述的用於僅查找相似於UPPERCASE參數加載配置的方法。

9.2 加載配置

有幾種方式加載配置。

9.2.1 從環境變量加載

任何由SANIC_定義的變量都將應用於sanic配置。例如，設置SANIC_REQUEST_TIMEOUT自動加載應用程序。你可使用load_cars將布爾值傳遞給Sanic構造函數來進行覆蓋。

app = Sanic(load_vars=False)

9.2.2 從對象加載

若是有不少配置參數而且它們有合理的默認值，將它們放置於模塊是有幫助的。

import myapp.default_settings

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

你也可使用類或者其它的對象類型。

9.2.3 從文件加載

一般狀況下，你想要從文件中加載配置參數。你能夠從from_file(/path/to/config_file)來加載配置參數。然而，這須要程序知道配置文件的位置，因此你能夠在環境變量中指定配置文件的路徑，並讓Sanic尋找配置文件並使用配置文件。

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

而後你能夠在MYAPP_SETTINGS環境設置下運行你的應用程序：

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

配置文件是常規的Python文件，運行它們只是爲了加載配置。這容許你使用任何正確的邏輯進行正確的配置。只要uppercase變量被添加到配置中，最多見的配置包括簡單的鍵值對：

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

9.3 內置的配置參數

提供的幾個預設值能夠在建立應用程序的時候被覆蓋：

Variable	Default	Description
REQUEST_MAX_SIZE	100000000	How big a request may be (bytes)
REQUEST_TIMEOUT	60	How long a request can take (sec)
KEEP_ALIVE	True	Disables keep-alive when False

10. Cookies

Cookies是持續保存在用戶瀏覽器中的數據片斷。Sanic能夠讀取和寫入Cookies，並以鍵值對的形式保存。

10.1 讀取Cookies

能夠經過Request對象的cookies字典訪問訪問用戶的cookies。

from sanic.response import text

@app.route("/cookie")
async def test(request):
    test_cookie = request.cookies.get('test')
    return text("Test cookie set to: {}".format(test_cookie))

10.2 寫入Cookies

當返回一個響應時，能夠在Response對象上設置Cookies。

from sanic.response import text

@app.route("/cookie")
async def test(request):
    response = text("There's a cookie up in this response")
    response.cookies['test'] = 'It worked!'
    response.cookies['test']['domain'] = '.gotta-go-fast.com'
    response.cookies['test']['httponly'] = True
    return response

10.3 刪除Cookies

能夠語義或明確地刪除Cookies。

from sanic.response import text

@app.route("/cookie")
async def test(request):
    response = text("Time to eat some cookies muahaha")

    # This cookie will be set to expire in 0 seconds
    del response.cookies['kill_me']

    # This cookie will self destruct in 5 seconds
    response.cookies['short_life'] = 'Glad to be here'
    response.cookies['short_life']['max-age'] = 5
    del response.cookies['favorite_color']

    # This cookie will remain unchanged
    response.cookies['favorite_color'] = 'blue'
    response.cookies['favorite_color'] = 'pink'
    del response.cookies['favorite_color']

    return response

響應的cookies能夠設置爲字典值，同時也有如下參數可用：

• expires(時間): cookie最後在客戶端瀏覽器上存在時間。
• path(字符串): Cookie的URL子集。默認爲/。
• comment(字符串): 註釋（元數據）。
• domain(字符串): 指定cookie有效的域。顯式指定的域必須始終以點開頭。
• max-age(數字): cookie應該存在的秒數。
• secure(布爾值): 指定cookie是否只能經過HTTPS發送。
• httponly(布爾值): 指定cookie是否能被Javascript讀取。

11. 處理器裝飾

因爲Sanic處理程序大都是簡單的Python函數，於是你能夠用相似Flask的方式對其進行裝飾。典型的例子就是在你的執行程序以前運行一些你想運行的代碼。

11.1 裝飾受權

假設你須要檢查用戶是否有權訪問特定的端點，你能夠建立一個包裝處理函數的裝飾起，若是客戶端有權訪問資源，則檢查請求，併發送適當的響應。

from functools import wraps
from sanic.response import json

def authorized():
    def decorator(f):
 @wraps(f)
        async def decorated_function(request, *args, **kwargs):
            # run some method that checks the request
            # for the client's authorization status
            is_authorized = check_request_for_authorization_status(request)

            if is_authorized:
                # the user is authorized.
                # run the handler method and return the response
                response = await f(request, *args, **kwargs)
                return response
            else:
                # the user is not authorized. 
                return json({'status': 'not_authorized'}, 403)
        return decorated_function
    return decorator


@app.route("/")
@authorized()
async def test(request):
    return json({status: 'authorized'})

12. 流媒體

12.1 請求流媒體

以下所示，Sanic容許你以流的方式請求數據。當請求結束的時候，``request.stream.get()將返回None`值。只有post、put和patch裝飾器有流的參數。

from sanic import Sanic
from sanic.views import CompositionView
from sanic.views import HTTPMethodView
from sanic.views import stream as stream_decorator
from sanic.blueprints import Blueprint
from sanic.response import stream, text

bp = Blueprint('blueprint_request_stream')
app = Sanic('request_stream')


class SimpleView(HTTPMethodView):

 @stream_decorator
    async def post(self, request):
        result = ''
        while True:
            body = await request.stream.get()
            if body is None:
                break
            result += body.decode('utf-8')
        return text(result)


@app.post('/stream', stream=True)
async def handler(request):
    async def streaming(response):
        while True:
            body = await request.stream.get()
            if body is None:
                break
            body = body.decode('utf-8').replace('1', 'A')
            response.write(body)
    return stream(streaming)


@bp.put('/bp_stream', stream=True)
async def bp_handler(request):
    result = ''
    while True:
        body = await request.stream.get()
        if body is None:
            break
        result += body.decode('utf-8').replace('1', 'A')
    return text(result)


async def post_handler(request):
    result = ''
    while True:
        body = await request.stream.get()
        if body is None:
            break
        result += body.decode('utf-8')
    return text(result)

app.blueprint(bp)
app.add_route(SimpleView.as_view(), '/method_view')
view = CompositionView()
view.add(['POST'], post_handler, stream=True)
app.add_route(view, '/composition_view')


if __name__ == '__main__':
    app.run(host='127.0.0.1', port=8000)

12.2 響應流媒體

Sanic容許你使用流的方法將內容以流的方式傳輸到客戶端。當傳遞一個能夠寫入的StreamingHTTPResponse對象時，這個方法接受協程回調（coroutine callback）。一個簡單的例子以下：

from sanic import Sanic
from sanic.response import stream

app = Sanic(__name__)

@app.route("/")
async def test(request):
    async def sample_streaming_fn(response):
        response.write('foo,')
        response.write('bar')

    return stream(sample_streaming_fn, content_type='text/csv')

在你想將以流的方式傳遞內容到外部服務的客戶端（如數據庫）的時候頗有用。例如，你可使用asyncpg提供的異步遊標將數據庫的記錄以流的方式傳遞到客戶端。

@app.route("/")
async def index(request):
    async def stream_from_db(response):
        conn = await asyncpg.connect(database='test')
        async with conn.transaction():
            async for record in conn.cursor('SELECT generate_series(0, 10)'):
                response.write(record[0])

    return stream(stream_from_db)

13. 基於類的視圖

基於類的視圖只是爲了實現對響應行爲的請求的簡單類。它們提供了在同一端點對不一樣HTTP請求類型進行區分處理的方法。

端點能夠分配一個基於類的視圖，而不是定義和裝飾三種不一樣的處理函數和一個用於每一個端點的請求類型。

13.1 定義視圖

基於類的視圖是HTTPMethodView的子類。你能夠爲每一個HTTP請求實現你想要的類方法。若是一個請求沒有定義方法，一個405:Method not allowed的響應就會生成。

要在端點上註冊基於類的視圖，就須要使用app.add_route方法。它的第一個參數是as_view方法定義的類，第二個參數是URL端點。

可被使用的方法包括get,post,patch,put和delete方法。使用這些方法的方式以下所示：

from sanic import Sanic
from sanic.views import HTTPMethodView
from sanic.response import text

app = Sanic('some_name')

class SimpleView(HTTPMethodView):

  def get(self, request):
      return text('I am get method')

  def post(self, request):
      return text('I am post method')

  def put(self, request):
      return text('I am put method')

  def patch(self, request):
      return text('I am patch method')

  def delete(self, request):
      return text('I am delete method')

app.add_route(SimpleView.as_view(), '/')

你也可使用異步async語法。

from sanic import Sanic
from sanic.views import HTTPMethodView
from sanic.response import text

app = Sanic('some_name')

class SimpleAsyncView(HTTPMethodView):

  async def get(self, request):
      return text('I am async get method')

app.add_route(SimpleAsyncView.as_view(), '/')

13.2 URL 參數

若是你須要任何URL參數，就像路由章節介紹的同樣，將其包含在方法定義中。

class NameView(HTTPMethodView):

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

app.add_route(NameView.as_view(), '/<name>')

13.3 裝飾

若是你想添加任何裝飾器到類中，能夠設置decorators類變量。當調用as_view方法的時候，會應用於類中。

class ViewWithDecorator(HTTPMethodView):
  decorators = [some_decorator_here]

  def get(self, request, name):
    return text('Hello I have a decorator')

app.add_route(ViewWithDecorator.as_view(), '/url')

13.4 構造URL

若是你但願爲HTTPMethodView構造一個URL，類的名字將會看成url_for的端點。以下所示：

@app.route('/')
def index(request):
    url = app.url_for('SpecialClassView')
    return redirect(url)


class SpecialClassView(HTTPMethodView):
    def get(self, request):
        return text('Hello from the Special Class View!')


app.add_route(SpecialClassView.as_view(), '/special_class_view')

13.5 使用組成視圖(CompositionView)

作爲HTTPMethodView的替代方法，你能夠在視圖類外使用CompositionView來移動處理函數。

處理函數在來源的每一個HTTP方法中都進行了定義，而後使用CompositionView.add方法來添加視圖。
它的第一個參數應該是一個HTTP處理方法的列表(如['GET', 'POST'])，第二個參數是處理函數。
下面的例子展現瞭如何在CompositionView中使用外部處理函數和內聯lanbda方法。

from sanic import Sanic
from sanic.views import CompositionView
from sanic.response import text

app = Sanic(__name__)

def get_handler(request):
    return text('I am a get method')

view = CompositionView()
view.add(['GET'], get_handler)
view.add(['POST', 'PUT'], lambda request: text('I am a post/put method'))

# Use the new view to handle requests to the base URL
app.add_route(view, '/')

須要注意的是，當前你沒法使用url_for方法爲CompositionView構建一個URL。

14. 定製協議

你能夠經過自定義一個協議來更改Sanic已經定義的協議行爲，這個協議是asyncio.protocol的子類。這個協議可以以protocol關鍵字參數傳遞給sanic.run方法。

自定義協議類的構造函數接受如下的關鍵字參數：

• loop: asyncio兼容的事件循環；

• connections: 存儲協議的set對象。當Sanic接收SIGINT和SIGTERM參數時，它會對集合中的全部協議對象執行protocol.close_if_idle。

• signal: 一個具備stop屬性的sanic.server.Signal對象。當Sanic接收到SIGINT和SIGTERM參數的時候，signal.stopped將被賦值爲True。

• request_handler: 一個將sanic.request.Request對象和response回調爲參數的協程程序。

• error_handler: 一個當出現異常時被調用出的sanic.exceptions.Handler對象。

• request_timeout: 請求超時前的秒數。

• request_max_size: 指定最大的請求數，以字節爲單位。

14.1 例子

若是處理函數沒有返回HTTPResponse對象，那沒在默認的協議中就會發生錯誤。

經過複寫write_response協議方法，若是處理程序返回一個字符串，它將被轉換爲一個HTTPResponse對象。

from sanic import Sanic
from sanic.server import HttpProtocol
from sanic.response import text

app = Sanic(__name__)


class CustomHttpProtocol(HttpProtocol):

    def __init__(self, *, loop, request_handler, error_handler, signal, connections, request_timeout, request_max_size):
        super().__init__(
            loop=loop, request_handler=request_handler,
            error_handler=error_handler, signal=signal,
            connections=connections, request_timeout=request_timeout,
            request_max_size=request_max_size)

    def write_response(self, response):
        if isinstance(response, str):
            response = text(response)
        self.transport.write(
            response.output(self.request.version)
        )
        self.transport.close()


@app.route('/')
async def string(request):
    return 'string'


@app.route('/1')
async def response(request):
    return text('response')

app.run(host='0.0.0.0', port=8000, protocol=CustomHttpProtocol)

15. SSL範例

SSLContext能夠選擇進行傳遞。

import ssl
context = ssl.create_default_context(purpose=ssl.Purpose.CLIENT_AUTH)
context.load_cert_chain("/path/to/cert", keyfile="/path/to/keyfile")

app.run(host="0.0.0.0", port=8443, ssl=context)

你還能夠將證書和密鑰的位置作爲自檢進行傳遞。

ssl = {'cert': "/path/to/cert", 'key': "/path/to/keyfile"}
app.run(host="0.0.0.0", port=8443, ssl=ssl)

16. 日誌記錄

Sanic容許你使用python3 的logging API對請求不一樣類型日誌進行記錄（諸如訪問記錄，錯誤記錄）。

16.1 快速教程

下面是一個使用默認設置的簡單示例：

from sanic import Sanic
from sanic.config import LOGGING

# The default logging handlers are ['accessStream', 'errorStream']
# but we change it to use other handlers here for demo purpose
LOGGING['loggers']['network']['handlers'] = [
    'accessSysLog', 'errorSysLog']

app = Sanic('test')

@app.route('/')
async def test(request):
    return response.text('Hello World!')

if __name__ == "__main__":
  app.run(log_config=LOGGING)

若是須要關閉日誌記錄，只須要分配log_config=None:

if __name__ == "__main__":
  app.run(log_config=None)

這將在處理請求的時候跳過調用日誌的功能，你能夠加快在使用中的速度。

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

16.2 配置

默認狀況下，使用sanic.config.LOGGING字典來設置log_config參數，下面是handlers默認配置中設置的默認值：

• internal: 使用logging.StreamHandler 內部信息在控制檯輸出。

• accessStream: 使用logging.StreamHandler 登陸控制檯的請求信息。

• errorStream: 使用logging.StreamHandler 控制檯的錯誤信息和追溯信息。

• accessSysLog: 使用logging.handlers.SysLogHandler 記錄到syslog的請求信息。

• errorSysLog: 使用logging.handlers.SysLogHandler syslog的錯誤消息和追溯記錄。

filters過濾:

• accessFilter: 使用sanic.log.DefaultFilter 只容許DEBUG，INFO和NONE(0)級別的過濾器。

• errorFilter: 使用sanic.log.DefaultFilter 只容許在WARNING，ERROR和CRITICAL級別的過濾器。

sanic中使用了兩種loggers，若是要建立本身的日誌記錄配置，則必須對它們進行定義：

• sanic: 記錄內部信息。

• network: 記錄來自網絡請求，以及請求中的任何信息。

16.3 日誌格式

除了由python(asctime,levelname,message)提供的默認參數以外，Sanic還爲accessFilter提供了網絡記錄器的其餘參數：

• host (str) request.ip

• request (str) request.method + " " + request.url

• status (int) response.status

• byte (int) len(response.body)

默認訪問日誌格式爲:

%(asctime)s - (%(name)s)[%(levelname)s][%(host)s]: %(request)s %(message)s %(status)d %(byte)d

17. 測試

Sanic端點可使用test_client對象進行本地測試，test_client對象依賴與aiohttp庫。

該test_client展現瞭如何在你的app使用get,post,put,delete,patch,head和options方法。一個簡單的例子(使用pytest)以下：

# Import the Sanic app, usually created with Sanic(__name__)
from external_server import app

def test_index_returns_200():
    request, response = app.test_client.get('/')
    assert response.status == 200

def test_index_put_not_allowed():
    request, response = app.test_client.put('/')
    assert response.status == 405

在內部每調用一次test_client方法的時候，運行在127.0.01:42101的Sanic的app的測試請求將使用aiohttp執行測試。

test_client方法接受如下的參數和關鍵字參數：

• uri(默認'/'): 表示要測試的URL字符串。

• gather_request(默認True): 布爾值，用於肯定該函數是否返回原始請求，若是設置爲True，則返回值是一個元組(request, response)，若是爲False則只返回響應。

• server_kwargs(默認爲{}): 在運行測試請求以前傳遞給app.run的附加參數。

• debug(默認False): 布爾值，用戶肯定是否在調試模式下運行該服務。

函數還將(request_args)和(*request_kwargs)直接傳遞給aiohttp ClientSession進行請求。

例如，要向GET請求中提供數據，能夠執行一下的操做：

def test_get_request_includes_data():
    params = {'key1': 'value1', 'key2': 'value2'}
    request, response = app.test_client.get('/', params=params)
    assert request.args.get('key1') == 'value1'

還有提供數據到 JSON POST請求中：

def test_post_json_request_includes_data():
    data = {'key1': 'value1', 'key2': 'value2'}
    request, response = app.test_client.post('/', data=json.dumps(data))
    assert request.json.get('key1') == 'value1'

17.1 pytest-sanic

pytest-sanic是一個pytest插件，它能夠幫助您異步測試代碼。

就像下面這樣進行測試：

async def test_sanic_db_find_by_id(app):
    """ Let's assume that, in db we have, { "id": "123", "name": "Kobe Bryant", "team": "Lakers", } """
    doc = await app.db["players"].find_by_id("123")
    assert doc.name == "Kobe Bryant"
    assert doc.team == "Lakers"

pytest-sanic還提供了一些有用的設置，如loop，unused_port，test_server，test_client。

@pytest.yield_fixture
def app():
    app = Sanic("test_sanic_app")

 @app.route("/test_get", methods=['GET'])
    async def test_get(request):
        return response.json({"GET": True})

 @app.route("/test_post", methods=['POST'])
    async def test_post(request):
        return response.json({"POST": True})

    yield app


@pytest.fixture
def test_cli(loop, app, test_client):
    return loop.run_until_complete(test_client(app, protocol=WebSocketProtocol))


#########
# Tests #
#########

async def test_fixture_test_client_get(test_cli):
    """ GET request """
    resp = await test_cli.get('/test_get')
    assert resp.status == 200
    resp_json = await resp.json()
    assert resp_json == {"GET": True}

async def test_fixture_test_client_post(test_cli):
    """ POST request """
    resp = await test_cli.post('/test_post')
    assert resp.status == 200
    resp_json = await resp.json()
    assert resp_json == {"POST": True}

18. 部署

內建的Web服務器簡化了Sanic的部署。在定義了一個sanic.Sanic的實例後，在調用run方法可使用如下關鍵字參數：

• host(默認「127.0.0.1」): 服務器主機的地址。

• port(默認8000): 服務器的端口。

• debug(默認False): 啓用調試（減慢服務器速度）。

• ssl(默認None): 用於工做者SSL加密的SSLContext。

• sock（默認None）：服務器接受鏈接的Socket。

• worker（默認值1）：生成的工做進程數。

• loop（默認None）: asyncio兼容的事件循環。若是沒有指定，Sanic會建立本身的事件循環。

• protocol（默認HttpProtocol）：asyncio.protocol的子類。

18.1 進程

默認狀況下，Sanic在主進程中只偵聽一個CPU內核。要啓動其它核心，只需指定run參數中進程的數量。

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

Sanic將自動啓動多個進程並在它們之間創建路由路徑。建議進程數和CPU核心數同樣。

18.2 經過命令行運行

若是你喜歡使用命令行參數，則能夠經過執行模塊啓動Sanic服務器。例如，若是你將Sanic應用程序在名爲server.py的文件中初始化，那麼能夠像這樣運行服務：

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

使用這種運行sanic的方法，沒有必要在你的Python文件中調用app.run。若是須要這樣作，請確保包裝它，以便它由解釋器直接執行。

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

18.3 經過Gunicorn運行

Gunicorn'Green Unicorn'是用於UNIX的WSGI HTTP服務。

使用Gunicorn運行Sanic應用程序，您須要使用特殊的sanic.worker.GunicornWorker對象定義Gunicornworker-class參數：

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

若是您的應用程序遇到內存泄漏，您能夠配置Gunicorn在處理給定數量的請求後，正常地從新啓動一個工做。這能夠幫助限制內存泄漏的影響。

18.4 異步支持

異步支持合適與其餘應用程序（特別是loop）共享sanic進程。可是請注意，由於此方法不支持使用多個進程，通常不是運行應用程序的首選方式。

下面是一個不完整的例子：

server = app.create_server(host="0.0.0.0", port=8000)
loop = asyncio.get_event_loop()
task = asyncio.ensure_future(server)
loop.run_forever()

19. 擴展

• Sessions: session的支持，容許使用redis，memcache或內存進行存儲。

• CORS: 用於處理跨域資源共享的擴展。

• Compress: 容許您輕鬆地壓縮Sanic響應。

• Jinja2: Jinja2模板框架。

• OpenAPI/Swagger：OpenAPI支持，以及Swagger UI。

• Pagination: 簡單的分頁支持。

• Motor: Simple motor wrapper。

• Sanic CRUD：基於peewee 模型的CRUD(建立/檢索/更新/刪除)REST API自動生成的框架。

• UserAgent: 添加user_agent到請求

• Limiter: 限制sanic速率。

• Sanic EnvConfig：將環境變量加入sanic配置。

• Babel：藉助Babel庫，向Sanic應用程序添加i18n/l10n支持。

• Dispatch: 由werkzeug的DispatcherMiddleware驅動的調度程序。能夠做爲Sanic-to-WSGI適配器。

• Sanic-OAuth: 用於鏈接和建立本身的token受權的庫。

• Sanic-nginx-docker-example: 在nginx使用docker-compose的一個簡單易用的Sanic例子。

• sanic-graphql: Sanic的GraphQL集成。

• sanic-prometheus: Sanic的Prometheus指標。

• Sanic-RestPlus: Sanic的Flask-RestPlus端口。基於SwaggerUI的全功能REST API。

• sanic-transmute: 可從python函數和類生成API，並自動生成Swagger UI文檔。

• pytest-sanic: 一個用於Sanic的pytest插件。能夠測試異步代碼。

• jinja2-sanic：一個用於Sanic的jinja2模板渲染器。

20. API參考

http://sanic.readthedocs.io/en/latest/sanic/api_reference.html