1. 程式人生 > >sanic官方文檔解析之Exception和Middleware,Listeners

sanic官方文檔解析之Exception和Middleware,Listeners

接受 類型 listener png 到來 bar sync def return

1,異常

技術分享圖片

異常是從處理請求內部拋出來的,並且通過Sanic自動的被處理異常,,異常用第一個參數攜帶異常信息,還可以接受在HTTP響應中要傳遞回的狀態代碼。引發異常

  • 1.1引發異常

技術分享圖片

自動觸發異常,,簡單的額使用raise從sanic.exceptions的模塊中拋出相關的異常即可

 技術分享圖片

你也可以使用中止函數的狀態碼,來拋異常 ,如上圖所示:

  • 1.2異常處理

技術分享圖片

@app.exception被使用來重寫處理默認的異常,,這個裝飾器期望一系列的異常作為參數去處理,你可以通過SanicException去抓住異常,裝飾異常處理函數必須攜帶請求和異常對象在參數中.

技術分享圖片

你也可以增加異常處理.

技術分享圖片

有些情況下,,你也許想添加一些更詳細的錯誤信息處理函數,可以默認提供,在那種情況下,你可以對Sanic的錯誤語法進行子類的劃分.  

  • 1.3有用的例外

技術分享圖片

一些有用的例外,目前如上所示:

  • NotFound:當找不到適合請求的路由時調用
  • ServerError:當服務器發生故障時調用。如果用戶代碼中出現異常,通常會發生這種情況
  • 有關引發異常 的完整列表,請參考sanic.exception模塊.

2, 中間件:

技術分享圖片

中間件是用在請求到來之前和響應請求之後,,中間件可以被用來修改 氫氣去和 響應,在中間件處理函數中

另外,Sanic提供監聽運行的代碼的多種要點在你懂得的應用中

技術分享圖片

Sanic中有2中類型的中間件,請求和響應,者兩種響應都用@app.middleware裝飾器來裝飾.用裝飾器的參數是字符串來代替request或者是response.

from sanic import Sanic
from sanic.response import text

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


@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")
  • 2.1修改請求和響應

技術分享圖片

中間件能夠修改給定的請求和響應的參數,只要中間件沒有返回請求或者響應

from sanic import Sanic

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


@app.middleware("request")
async def add_key(request):
    # 增加請求對象的關鍵字,就像地點的對象
    request["foo"] = "bar"


@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, debug=True)

技術分享圖片

以上代碼應用在中間件中,第一個中間件增加一個新的關鍵字foo到請求request中,中間件的運行是因為請求對象的主體類似字典對象,第二個中間件定制的橫幅將會改變HTTP響應頭為Fake-Server,最後一個中間件是為了增加HTTP頭防止xss攻擊,這些2個功能在函數返回一個 response響應的時候調用

  • 2.2早期的回應

技術分享圖片

如果中間件返回一個httpresponse對象,那麽請求將停止處理並返回響應。如果在到達相關的用戶路由處理程序之前請求出現這種情況,則永遠不會調用該處理程序。返回響應還將阻止任何其他中間件的運行

from sanic import Sanic
from sanic.response import text

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


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

3,監聽

技術分享圖片

如你想執行啟動/拆卸的代碼作為你服務器的啟動或者關閉,你可以使用以下幾種監聽模式:

  • before_server_start
  • after_server_start
  • before_server_stop
  • after_server_stop

這些監聽的方式被作為裝飾在接收app項目也異步循環的功能函數的裝飾器

from sanic import Sanic

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


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

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

技術分享圖片

監聽同樣也可以用register_listener方法來註冊一個監聽者,如果你定義你的監聽者在另一個模塊中,此外在你實例化你的app中.

from sanic import Sanic

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


async def setup_db(app, loop):
    app.db = await db_setup()

app.register_listener(setup_db, "before_server_start")  # 註冊監聽者(在服務啟動前) 

技術分享圖片

如果要計劃在循環啟動後運行後臺任務,Sanic提供了add_task方法很容易啟動後臺程序.

from sanic import Sanic
import asyncio

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

async def notify_server_started_after_five_second():
    await asyncio.sleep(5)
    print("Server successfully started!")
# 用app來增加異步功能函數的任務
app.add_task(notify_server_started_after_five_second())

技術分享圖片

Sanic將會嘗試去自動的 諸如到app中,作為一個參數添加到任務中.

import asyncio
from sanic import Sanic


# 實例化一個Sanic對象
app = Sanic()
async def notify_server_started_after_five_second(app):
    await asyncio.sleep(5)
    print(app.name)
# 用app來增加異步函數功能
app.add_task(notify_server_started_after_five_second)

技術分享圖片

或者你可以明確地作用於app,也是一樣的效果.

from sanic import Sanic
import asyncio

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


async def notify_server_started_after_five_seconds(app):
    await asyncio.sleep(5)
    print(app.name)
app.add_task(notify_server_started_after_five_seconds(app))

sanic官方文檔解析之Exception和Middleware,Listeners