1,Custom Protocol:自定義協議

溫馨提示:自定義協議是一個高級用法,大多數的讀者不需要用到此功能

通過特殊的自定義協議,你可以改變sanic的協議,自定義協議需要繼承子類asyncio.protocol,這個子類在sanic.run方法中傳輸關鍵字protocol協議

自定義協議的構造類從sanic中接收關鍵字參數.

• loop: an asyncio-compatible event loop.(循環:異步兼容事件循環)
• connections: a set to store protocol objects. When Sanic receives SIGINT or SIGTERM
  , it executes protocol.close_if_idle for all protocol objects stored in this set.(連接:一個存儲協議對象的集合當接收到一個信號的時候,將會執行自定義的協議,close_if_idle會為所有的協議對象存儲在這個集合中)
• signal: a sanic.server.Signal object with the stopped attribute. When Sanic receives SIGINTor SIGTERM, signal.stopped is assigned True(信號:一個sanic服務器信號對象是用sanic停止的屬性來控制的,當sanic接收到信號,stopped的屬性就會變成Ture)
• request_handler: a coroutine that takes a sanic.request.Request object and a response callback as arguments(處理請求:將sanic的請求對象和響應回調作為參數的協程)
• error_handler: a sanic.exceptions.Handler which is called when exceptions are raised.(錯誤處理程序:當異常被拋出的時候,調用sanic.exception.Handler)
• request_timeout: the number of seconds before a request times out(請求超時:在請求超時前會有一些秒數.)
• request_max_size: an integer specifying the maximum size of a request, in bytes.(最大請求的量:指定請求最大的整數,以字節為單位)

例子:

如果一個處理函數沒有返回一個HTTPResponse對象,表名一個錯誤的發生在默認的謝一中.通過重寫write

_response協議方法,如果處理程序返回一個字符串被轉換成一個HTTPResponse響應對象

from sanic import Sanic
from sanic.server import HttpProtocol  # 在sanic的服務中存在HTTP協議
from sanic.response import text

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


class CustomHttpProtocol(HttpProtocol):

    # 初始化方法
    def __init__(
            self,
            *,
            loop,
            request_handdler,
            error_handler,
            signal,
            connections,
            request_timeout,
            request_max_size,
    ):
        # 執行父類的初始化方法
        super().__init__(
            loop=loop,
            request_handler=request_handdler,
            error_handler=error_handler,
            signal=signal,
            connections=connections,
            request_timeout=request_timeout,
            request_max_size=request_max_size
        )

    # 返回響應的方法
    def write_response(self, response):
        # 如果存在response對象
        if isinstance(response, str):
            # 封裝response響應對象
            response = text(response)
        self.transport.write(
            # 封裝請求的信息
            response.output(self.request.version)
        )
        self.transport.close()


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


@app.route("/i")
async def response(request):
    return text("response")

app.run(host="0.0.0.0", port=8000, protocol=CustomHttpProtocol)  # 啟動服務的時候指定協議為自定義協議

2,Socket(網絡套接字)

Sanic可以使用Python中的socket模塊來容納IPV4的socket

from sanic import Sanic
from sanic.response import json
import socket
# 實例化socket對象
sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
sock.bind(("::", 7777))  # 綁定一個元組
# 實例化一個sanic對象
app = Sanic()


# 業務處理函數
@app.route("/")
async def test(request):
    return json({"hello": "world"})

if __name__ == ‘__main__‘:
    app.run(sock=sock)  # 以網絡套接字的形式開啟服務

UNIX的例子如下:

import signal
import sys
import socket
import os
from sanic import Sanic
from sanic.response import json

server_socket = "/tmp/sanic.sock"
# 實例化socket對象
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)  # socket.UNIX這個參數不對,是不是官網錯了
sock.bind(server_socket)  # 綁定
# 實例化Sanic對象
app = Sanic()


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


def signal_handler(sig, frame):
    print("Exiting")
    os.unlink(server_socket)
    sys.exit(0)

if __name__ == ‘__main__‘:
    app.run(sock=sock)

sanic官方文檔解析之Custom Protocols(自定義協議)和Socket(網絡套接字)