1,示例

這部的文檔是簡單的示例集合,它能夠幫助你快速的啟動應用大部分的應用,這些應用大多事分類的,並且提供給ini工作的連接代碼:

1.1,基礎示例

這部分示例集成了提供簡單sanic簡單的代碼

單一APP

一個簡單的sanic應用with一個簡單的異步方法通過text和json類型的響應.

from sanic import Sanic
from sanic import response as res
# 實例化Sanic對象
app = Sanic(__name__)


# 服務器訪問的路由
@app.route("/")
async def test(req):
    return res.text("I am a teapot", status=418)  # 返回的是字符串格式的數據

# 開啟服務
if __name__ == ‘__main__‘:
    app.run(host="0.0.0.0", port=8000, debug=True)
 

from sanic import Sanic
from sanic import response

# 實例化Sanic對象
app = Sanic(__name__)


# 服務器的訪問路由
@app.route("/")
async def test(request):
    return response.json({"test": True})  # 返回的是json數據格式

# 啟動服務
if __name__ == ‘__main__‘:
    app.run(host="0.0.0.0", port=8000)

1.2,使用sanic.view來創建app應用

展示了使用sanic.viewes.httpmethodview的簡單機制，以及將其擴展為為為視圖提供自定義異步行為的方法

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

# 實例化Sanic對象
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, requets):
        return text("I am patch method")

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


class SimpleAsyncView(HTTPMethodView):
    async def get(self, request):
        return text("I am async get method")
    
    async def post(self, request):
        return text("I am async post method")
    
    async def put(self, request):
        return text("I am async put method")
    
    
app.add_route(SimpleView.as_view(), "/")  # 註冊視圖類(後邊加訪問的路由)
app.add_route(SimpleAsyncView.as_view(), "/async")  # 註冊視圖類(後邊加訪問的路由)

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

1.3,URL跳轉

from sanic import Sanic
from sanic import response

app = Sanic(__name__)


# 同步函數的跳轉
@app.route("/")
def handle_response(request):
    return response.redirect("/redirect")


# 異步處理函數到的跳轉
@app.route("/redirect")
async def test(request):
    return response.json({"Redirected": True})

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

1.4,redirection URL

Sanic提供了一種跳轉頁面的簡單方法叫作url_for,它將會獨一無二的url名字作為參數並且返回給你分配實際的路由地址.,有了url_for這個方法將幫助我們簡化在不同應用跳轉中請求的影響,

from sanic import Sanic
from sanic import response

app = Sanic(__name__)


@app.route("/")
async def index(request):
    # 用post_handler生成url端
    url = app.url_for("post_handler", post_id=5)
    # 提升成的url最後跳轉的頁面變成了"/post/5"
    return response.redirect(url)


@app.route("/posts/<post_id>")
async def post_handler(request, post_id):
    return response.text("Post- {}".format(post_id))
                         
if __name__ == ‘__main__‘:
    app.run(host="0.0.0.0", port=8000, debug=True)

1.5,藍圖

Sanic提供了一個驚人的特性，可以將您的API和路由分組到一個邏輯集合中，這個集合可以很容易地導入並插入到您的任何Sanic應用程序中，它被稱為藍圖。

from sanic import Blueprint, Sanic
from sanic.response import file,json
# 實例化Sanic對象
app = Sanic(__name__)
blueprint = Blueprint("name", url_prefix="/my_blueprint")
blueprint2 = Blueprint("name2", url_prefix="/my_blueprint2")
blueprint3 = Blueprint("name3", url_prefix="/my_blueprint3")


# blueprint藍圖的路由
@blueprint.route("/foo")
async def foo(request):
    return json({"msg": "hi from blueprint"})


# blueprint2藍圖的路由
@blueprint2.route("/foo")
async def foo2(request):
    return json({"msg": "hi from blueprint2"})


# blueprint3藍圖的路由
@blueprint3.route("/foo")
async def index(request):
    return await file("websocket.html")


# websocket的路由
@app.websocket("/foo")
async def foo3(request, ws):
    while True:
        data = "hello!"
        print("Sending:" + data)
        await ws.send(data)
        data = await ws.recv()
        print("Received:" + data)
# 註冊藍圖
app.blueprint(blueprint)  
app.blueprint(blueprint2)
app.blueprint(blueprint3)

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

1.6,日誌功能增強

from sanic import Sanic
from sanic import response
import logging

logging_format = "[%(asctime)s] % (process)d-%(Levelname)s"
logging_format += "%(module)s::%(funcName)s():1%(Lineno)d:"
logging_format += "%(message)s"

logging.basicConfig(
    format=logging_format,
    level=logging.DEBUG
)
log = logging.getLogger()

# 設置日誌去重寫默認的配置
sanic = Sanic()


@sanic.route("/")
def test(request):
    log.info("received request; responding with ‘hey‘")
    return response.text("hey")


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

下面的示例代碼演示了sanic.app.sanic.middleware（）的用法，以便提供一種機制為每個傳入請求分配一個唯一的請求ID，並通過aiotask上下文記錄它們

"""
基於https://github.com/skyscanner/aiotask-context中的示例
和“examples/override logging，run.py”。
需要https://github.com/skyscanner/aiotask-context/tree/52efbc21e2def2d52abb9a8e951f3ce5e6f690或更高版本
$pip安裝git+https://github.com/skyscanner/aiotask-context.git
"""
import asyncio
import uuid
import logging
from signal import signal, SIGINT
from sanic import Sanic
from sanic import response
import uvloop  # windows下不支持uvloop
import aiotask_content as context

log = logging.getLogger(__name__)


class RequestIdFilter(logging.Filter):
    def filter(self, record):
        record.request_id = context.get("X-Request-ID")
        return True


LOG_SETTINGS = {
    ‘version‘: 1,
    ‘disable_existing_loggers‘: False,
    ‘handlers‘: {
        ‘console‘: {
            ‘class‘: ‘logging.StreamHandler‘,
            ‘level‘: ‘DEBUG‘,
            ‘formatter‘: ‘default‘,
            ‘filters‘: [‘requestid‘],
        },
    },
    ‘filters‘: {
        ‘requestid‘: {
            ‘()‘: RequestIdFilter,
        },
    },
    ‘formatters‘: {
        ‘default‘: {
            ‘format‘: ‘%(asctime)s %(levelname)s %(name)s:%(lineno)d %(request_id)s | %(message)s‘,
        },
    },
    ‘loggers‘: {
        ‘‘: {
            ‘level‘: ‘DEBUG‘,
            ‘handlers‘: [‘console‘],
            ‘propagate‘: True
        },
    }
}
app = Sanic(__name__, log_config=LOG_SETTINGS)
@app.route("/")
async def test(request):
    log.debug("X-Request-ID: %s", context.get("X-Request-ID"))
    log.info("Hello from test!")
    return response.json({"test": True})

if __name__ == ‘__main__‘:
    asyncio.set_event_loop(uvloop.new_event_loop())
    server = app.create_server(host="0.0.0.0", port=8000)
    loop = asyncio.get_event_loop()
    loop.set_task_factory(context.task_factory)
    task = asyncio.ensure_future(server)
    try:
        loop.run_forever()
    except:
        loop.stop()

1.7,sanic流支持

Sanic框架內置了對大文件流的支持，下面的代碼解釋了使用流支持設置SANIC應用程序的過程。

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 1:
            body = await request.stream.get()
            if not body:
                break
            result += body.decode("utf-8")
        return text(result)
    
    
@app.post("/stream", stream=True)
async def handler(request):
    async def streaming(response):
        while 1:
            body = await request.stream.get()
            if not body:
                break
            body = body.decode("utf-8").replace("1", "A")
            await response.write(body)
    return stream(streaming)


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


async def post_handler(request):
    result = ""
    while 1:
        body = await request.stream.get()
        if not body:
            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="0.0.0.0", port=8000)

簡單的客戶端展示使用客戶端代碼流應用(測試一下代碼)

import requests
# 警告:這是一個繁重的進程
data = ""
for i in range(1, 250000):
    data += str(i)
    
r = requests.post("http://0.0.0.0:8000/stream", data=data)

print(r.text)

1.8,sanic的並發支持

SANIC支持通過多個工作人員支持啟動應用程序。但是，為了確保高效的執行，能夠限制每個進程/循環的並發性是很重要的。下面的代碼部分提供了一個簡單的示例，說明如何在asyncio.semaphore的幫助下限制並發性

from sanic import Sanic
from sanic.response import json

import asyncio
import aiohttp
app = Sanic(__name__)

sem = None


@app.listener("before_server_start")
def init(sanic, loop):
    global sem
    concurrency_per_work = 4
    sem = asyncio.Semaphore(concurrency_per_work,loop=loop)


async def bounded_fetch(session, url):
    # 使用session對象執行url的get請求
    async with sem, session.get(url) as response:
        return await response.json()
    
    
@app.route("/")
async def test(request):
    # 下載並且提供json例子的服務
    url = "http://api.github.com/repos/channelcat/sanic"
    
    async with aiohttp.ClientSession() as session:
        response = await bounded_fetch(session, url)
        return json(response)
    
app.run(host="0.0.0.0", port=8000, workers=2)

sanic官方文檔解析之Example(一)