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

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)