乡下人产国偷v产偷v自拍,国产午夜片在线观看,婷婷成人亚洲综合国产麻豆,久久综合给合久久狠狠狠9

  • <output id="e9wm2"></output>
    <s id="e9wm2"><nobr id="e9wm2"><ins id="e9wm2"></ins></nobr></s>

    • 分享

      sanic異步框架之中文文檔

       尹培西 2021-04-11

      typora-copy-images-to: ipic


      [TOC]

      快速開始

      在安裝Sanic之前,讓我們一起來看看Python在支持異步的過程中,都經(jīng)歷了哪些比較重大的更新。

      首先是Python3.4版本引入了asyncio,這讓Python有了支持異步IO的標(biāo)準(zhǔn)庫,而后3.5版本又提供了兩個(gè)新的關(guān)鍵字async/await,目的是為了更好地標(biāo)識(shí)異步IO,讓異步編程看起來更加友好,最后3.6版本更進(jìn)一步,推出了穩(wěn)定版的asyncio,從這一系列的更新可以看出,Python社區(qū)正邁著堅(jiān)定且穩(wěn)重的步伐向異步編程靠近。

      安裝

      Sanic是一個(gè)支持 async/await 語法的異步無阻塞框架,這意味著我們可以依靠其處理異步請求的新特性來提升服務(wù)性能,如果你有Flask框架的使用經(jīng)驗(yàn),那么你可以迅速地使用Sanic來構(gòu)建出心中想要的應(yīng)用,并且性能會(huì)提升不少,我將同一服務(wù)分別用Flask和Sanic編寫,再將壓測的結(jié)果進(jìn)行對比,發(fā)現(xiàn)Sanic編寫的服務(wù)大概是Falsk的1.5倍。

      僅僅是Sanic的異步特性就讓它的速度得到這么大的提升么?是的,但這個(gè)答案并不標(biāo)準(zhǔn),更為關(guān)鍵的是Sanic使用了uvloop作為asyncio的事件循環(huán),uvloop由Cython編寫,它的出現(xiàn)讓asyncio更快,快到什么程度?這篇文章中有介紹,其中提出速度至少比 nodejs、gevent 和其他Python異步框架要快兩倍,并且性能接近于用Go編寫的程序,順便一提,Sanic的作者就是受這篇文章影響,這才有了Sanic。

      怎么樣?有沒有激起你學(xué)習(xí)Sanic的興趣,如果有,就讓我們一起開始學(xué)習(xí)吧,在開始之前,你只需要有一臺(tái)安裝了Python的電腦即可。

      說明:由于Windows下暫不支持安裝uvloop,故在此建議使用Mac或Linux

      虛擬環(huán)境

      程序世界一部分是對應(yīng)著現(xiàn)實(shí)的,在生活中,我們會(huì)在不同的環(huán)境完成不同的任務(wù),比如在廚房做飯、臥室休息,分工極其明確。

      其實(shí)用Python編寫應(yīng)用服務(wù)也是如此,它們同樣希望應(yīng)用服務(wù)與開發(fā)環(huán)境是一對一的關(guān)系,這樣做的好處在于,每個(gè)獨(dú)立的環(huán)境都可以簡潔高效地管理自身對應(yīng)服務(wù)所依賴的第三方庫,如若不然,各個(gè)服務(wù)都安排在同一環(huán)境,這樣不僅會(huì)造成管理上的麻煩,還會(huì)使第三方庫之間產(chǎn)生沖突。

      通過上面的敘述,我們是不是可以得出這樣一個(gè)核心觀點(diǎn):應(yīng)該在不同的環(huán)境下做不同的事 ,以此類推,寫項(xiàng)目的時(shí)候,我們也需要為每個(gè)不同的項(xiàng)目構(gòu)建一個(gè)無干擾的的環(huán)境,發(fā)散思維,總結(jié)一下:

      不同的項(xiàng)目,需要為其構(gòu)建不同的虛擬環(huán)境,以免互相干擾

      構(gòu)建虛擬環(huán)境的工具很多,如下:

      …...

      以上三個(gè)工具都可以快速地幫助我們構(gòu)建當(dāng)前需要的Python環(huán)境,如果你之前沒有使用過,可直接點(diǎn)開鏈接進(jìn)行下載,如果你正在使用其它的環(huán)境管理工具,也不要緊,因?yàn)椴徽撃闶褂媚囊环N方式,我們最終目的都是針對一個(gè)新項(xiàng)目構(gòu)建一個(gè)新的環(huán)境。

      安裝配置好之后,簡單看看官方提供的使用方法,就可以開始了,比如我本機(jī)使用的是venv(python3.5以后官方推薦使用這個(gè)venv來管理虛擬環(huán)境),安裝完成后可以很方便地創(chuàng)建一個(gè)虛擬環(huán)境,比如這里使用Python3.6來作為本書項(xiàng)目的默認(rèn)環(huán)境:

      cd ~/ # 新建一個(gè)python3.6環(huán)境 python3 -m venv pyenv # 安裝好之后 輸入下面命令進(jìn)入名為python36的環(huán)境 cd pyenv/ source bin/activate # 查看版本 python -V

      若安裝速度比較慢,可以考慮換國內(nèi)源,比如 國內(nèi)鏡像 ,至于為什么選擇python3.6作為默認(rèn)環(huán)境,一是因?yàn)镾anic只支持Python3.5+,二則是我們構(gòu)建的項(xiàng)目最終是要在生產(chǎn)環(huán)境下運(yùn)行的,所以建議最好安裝Python3.6下穩(wěn)定版本的asyncio。

      安裝Sanic

      Python安裝第三方模塊都是利用pip工具進(jìn)行安裝,這里也不例外,首先進(jìn)入上一步我們新建的 python3.6 虛擬環(huán)境,然后安裝:

      # 安裝Sanic,請先使用 source activate python36 進(jìn)入虛擬環(huán)境
      pip install sanic
      # 如果不想使用uvloop和ujson 可以這樣安裝
      SANIC_NO_UVLOOP=true SANIC_NO_UJSON=true pip install sanic

      通過上面的命令,你就可以在 python3.6 虛擬環(huán)境中安裝Sanic以及其依賴的第三方庫了,若想查看Sanic是否已經(jīng)正確安裝,可以進(jìn)入終端下對應(yīng)的虛擬環(huán)境,啟動(dòng)Python解釋器,導(dǎo)入Sanic庫:

      python >>> >>> import sanic

      如果沒有出現(xiàn)錯(cuò)誤,就說明你已經(jīng)正確地安裝了Sanic,請繼續(xù)閱讀下一節(jié),了解下如何利用Sanic來構(gòu)建一個(gè)Web項(xiàng)目吧。

      開始

      我們將正式使用Sanic來構(gòu)建一個(gè)web項(xiàng)目,讓我們踏出第一步,利用Sanic來編寫一個(gè)返回Hello World!字符串的服務(wù)程序。

      新建一個(gè)文件夾sanicweb

      $ mkdir sanicweb
      $ cd sanicweb/
      $ pwd
      /Users/junxi/pyenv/sanicweb

      創(chuàng)建一個(gè)sanic例子,保存為 main.py :

      from sanic import Sanic from sanic.response import text app = Sanic() @app.route('/') async def index(request): return text('Hello World!') if __name__ == '__main__': app.run(host='0.0.0.0', port=9000)

      運(yùn)行main.py,然后訪問地址http://127.0.0.1:9000/

      $ curl -X GET http://127.0.0.1:9000/
      Hello World!

      這樣我們就完成了第一個(gè)sanic例子。

      接下來,你將逐漸地了解到Sanic的一些基本用法,如路由的構(gòu)建、接受請求數(shù)據(jù)以及返回響應(yīng)的內(nèi)容等。

      路由

      路由允許用戶為不同的URL端點(diǎn)指定處理程序函數(shù)。

      實(shí)例:

      from sanic.response import json @app.route('/') async def index(request): return json({ 'hello': 'world' })

      url http://server.url/ 被訪問(服務(wù)器的基本url),最終/被路由器匹配到處理程序函數(shù),測試,然后返回一個(gè)JSON對象。

      必須使用async def語法來定義Sanic處理函數(shù),因?yàn)樗鼈兪钱惒胶瘮?shù)。

      請求參數(shù)

      要指定一個(gè)參數(shù),可以用像這樣的角引號<PARAM>包圍它。請求參數(shù)將作為關(guān)鍵字參數(shù)傳遞給路線處理程序函數(shù)。

      實(shí)例:

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

      重啟服務(wù),輸入地址http://127.0.0.1:9000/tag/python進(jìn)行訪問

      $ curl -X GET http://127.0.0.1:9000/tag/python Tag - python

      為參數(shù)指定類型,在參數(shù)名后面添加(:類型)。如果參數(shù)不匹配指定的類型,Sanic將拋出一個(gè)不存在的異常,導(dǎo)致一個(gè)404頁面

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

      測試結(jié)果如下:

      $ curl -X GET http://127.0.0.1:9000/number/1 Integer - 1 $ curl -X GET http://127.0.0.1:9000/number/asds Error: Requested URL /number/asds not found $ curl -X GET http://127.0.0.1:9000/number/12.0 Number - 12.0 $ curl -X GET http://127.0.0.1:9000/person/junxi Person - junxi $ curl -X GET http://127.0.0.1:9000/person/123 Error: Requested URL /person/123 not found $ curl -X GET http://127.0.0.1:9000/folder/img Folder - img $ curl -X GET http://127.0.0.1:9000/folder/img1 Folder - img1 $ curl -X GET http://127.0.0.1:9000/folder/images Error: Requested URL /folder/images not found $ curl -X GET http://127.0.0.1:9000/folder/2018 Folder - 2018

      請求類型

      路由裝飾器接受一個(gè)可選的參數(shù),方法,它允許處理程序函數(shù)與列表中的任何HTTP方法一起工作。

      實(shí)例1:

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

      實(shí)例2:

      @app.post('/post2') async def post_handler(request): return text('POST request - {}'.format(request.json)) @app.get('/get2') async def get_handler(request): return text('GET request - {}'.format(request.args))

      測試結(jié)果:

      $ curl -X GET http://127.0.0.1:9000/get1?name=junxi
      GET request - {'name': ['junxi']}
      $ curl -X GET http://127.0.0.1:9000/get2?name=junxi
      GET request - {'name': ['junxi']}
      $ curl -H 'Content-type: application/json' -X POST -d '{'name':'junxi', 'gender':'male'}' http://127.0.0.1:9000/post1 
      POST request - {'name': 'junxi', 'gender': 'male'}
      $ curl -H 'Content-type: application/json' -X POST -d '{'name':'junxi', 'gender':'male'}' http://127.0.0.1:9000/post2
      POST request - {'name': 'junxi', 'gender': 'male'}

      增加路由

      實(shí)例:

      async def handler1(request): return text('ok') async def handler2(request, name): return text('Folder - {}'.format(name)) async def personal_handler2(request, name): return text('Person - {}'.format(name)) app.add_route(handler1, '/test1') app.add_route(handler2, '/folder2/<name>') app.add_route(personal_handler2, '/personal2/<name:[A-z]>', methods=['GET'])

      測試結(jié)果:

      $ curl -X GET http://127.0.0.1:9000/test1 
      ok
      $ curl -X GET http://127.0.0.1:9000/folder2/aaa
      Folder - aaa
      $ curl -X GET http://127.0.0.1:9000/personal2/A
      Person - A
      $ curl -X GET http://127.0.0.1:9000/personal2/a
      Person - a

      url_for

      Sanic提供了一個(gè)urlfor方法,根據(jù)處理程序方法名生成url。避免硬編碼url路徑到您的應(yīng)用程序

      實(shí)例:

      @app.router('/') async def index(request): url = app.url_for('post_handler', post_id=5) return redirect(url) @app.route('posts/<post_id>') async def post_handler(request, post_id): return text('Post - {}'.format(post_id))

      給url_for的關(guān)鍵字參數(shù)不是請求參數(shù),它將包含在URL的查詢字符串中。例如:

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

      所有有效的參數(shù)必須傳遞給url以便構(gòu)建一個(gè)URL。如果沒有提供一個(gè)參數(shù),或者一個(gè)參數(shù)與指定的類型不匹配,就會(huì)拋出一個(gè)URLBuildError

      可以將多值參數(shù)傳遞給url

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

      經(jīng)過測試訪問/我們會(huì)發(fā)現(xiàn),url跳轉(zhuǎn)到了/posts/5 ,并打印出來Post - 5 的結(jié)果。

      redirect是從sanic.response導(dǎo)入的方法,用于處理url的重定向。

      網(wǎng)絡(luò)套接字路由

      WebSocket routes

      websocket可以通過裝飾路由實(shí)現(xiàn)

      實(shí)例:

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

      另外,添加websocket路由方法可以代替裝飾器

      async def feed(request, ws): pass app.add_websocket_route(my_websocket_handler, '/feed')

      請求

      request

      常用類型

      當(dāng)一個(gè)端點(diǎn)收到一個(gè)HTTP請求時(shí),路由功能被傳遞給一個(gè) Request對象。

      以下變量可作為Request對象的屬性訪問:

      • json (any) - JSON body
      from sanic.response import json
      
      @app.route('/json')
      def post_json(request):
          return json({ 'received': True, 'message': request.json })
      • args(dict) - 查詢字符串變量。查詢字符串是類似于URL的部分?key1=value1&key2=value2。如果該URL被解析,則args字典將如下所示:{'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參數(shù)。對于之前的同一個(gè)URL ?key1=value1&key2=value2, raw_args字典看起來就像:{'key1': 'value1', 'key2': 'value2'}。

        ?

      • files(dictionary of File objects) - 具有名稱,正文和類型的文件列表
      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) - post表單變量。
      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(bytes) - 發(fā)送原始主體。無論內(nèi)容類型如何,該屬性都允許檢索請求的原始數(shù)據(jù)。
      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)
      • headers (dict) - 包含請求標(biāo)頭的不區(qū)分大小寫的字典。
      • ip (str) - 請求者的IP地址。
      • port (str) - 請求者的端口地址。
      • socket (tuple) - 請求者的(IP,端口)。
      • app - 對處理此請求的Sanic應(yīng)用程序?qū)ο蟮囊?。?dāng)模塊內(nèi)部的藍(lán)圖或其他處理程序無法訪問全局app對象時(shí),這非常有用。

        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:與請求關(guān)聯(lián)的URL方案:httphttps
      • host:與請求關(guān)聯(lián)的主機(jī): localhost:8080
      • path:請求的路徑: /posts/1/
      • query_string:請求的查詢字符串:foo=bar或一個(gè)空白字符串''
      • uri_template:匹配路由處理程序的模板: /posts/<id>/
      • token:授權(quán)標(biāo)頭(Authorization)的值: Basic YWRtaW46YWRtaW4=

      使用getgetlist訪問數(shù)據(jù)

      返回字典的請求屬性實(shí)際上會(huì)返回一個(gè)dict被調(diào)用的子類 RequestParameters。使用這個(gè)對象的關(guān)鍵區(qū)別在于getgetlist方法之間的區(qū)別。

      • get(key, default=None)按照正常操作,除了當(dāng)給定鍵的值是列表時(shí),只返回第一個(gè)項(xiàng)目。
      • getlist(key, default=None)正常操作,返回整個(gè)列表

      響應(yīng)

      response

      text

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

      HTML

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

      JSON

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

      File

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

      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')

      File Streaming

      對于大文件,文件和流的組合

      from sanic import response @app.route('/big_file.png') async def handle_request(request): return await response.file_stream('/srv/www/whatever.png')

      Redirect

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

      Raw

      沒有進(jìn)行編碼的響應(yīng)

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

      Modify headers or status

      要修改頭部或狀態(tài)代碼,將頭部或狀態(tài)參數(shù)傳遞給這些函數(shù)

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

      靜態(tài)文件

      static_files

      靜態(tài)文件和目錄,比如一個(gè)圖像文件,在Sanic注冊時(shí)使用。該方法使用一個(gè)端點(diǎn)URL和一個(gè)文件名。指定的文件將通過指定的端點(diǎn)訪問。

      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)

      Note:目前,您不能使用url構(gòu)建一個(gè)靜態(tài)文件的URL

      模版

      html templates編寫

      編寫web服務(wù),自然會(huì)涉及到html,sanic自帶有html函數(shù),但這并不能滿足有些需求,故引入jinja2迫在眉睫。使用方法也很簡單:

      # novels_blueprint.py片段
      from sanic import Blueprint
      from jinja2 import Environment, PackageLoader, select_autoescape
      
      # 初始化blueprint并定義靜態(tài)文件夾路徑
      bp = Blueprint('novels_blueprint')
      bp.static('/static', './static/novels')
      
      # jinjia2 config
      env = Environment(
          loader=PackageLoader('views.novels_blueprint', '../templates/novels'),
          autoescape=select_autoescape(['html', 'xml', 'tpl']))
      
      def template(tpl, **kwargs):
          template = env.get_template(tpl)
          return html(template.render(kwargs))
          
      @bp.route('/')
      async def index(request):
          return template('index.html', title='index')

      這樣,就實(shí)現(xiàn)了jinja2 模版的引入。

      異常

      Exceptions

      拋出異常

      要拋出異常,只需從sanic異常模塊中提出相應(yīng)的異常。

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

      也可以自定義狀態(tài)碼

      from sanic.exceptions import abort
      from sanic.response import text
      @app.route('/youshallnotpass')
      def no_no(request):
              abort(401)
              # this won't happen
              text('OK')

      處理異常

      Handling exceptions

      裝飾器一個(gè)異常列表作為參數(shù)來處理。你可以通過SanicException來捕獲它們!裝飾異常處理函數(shù)必須將請求和異常對象作為參數(shù)。

      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)) @app.exception(NotFound) def handle_404_redirect(request, exception): uri = app.url_for('index') return redirect(uri)

      有用的異常

      Useful exceptions

      常用

      • NotFound:在沒有找到合適的請求路徑時(shí)調(diào)用
      • ServerError:當(dāng)服務(wù)器內(nèi)部出現(xiàn)問題時(shí)調(diào)用。如果在用戶代碼中出現(xiàn)異常,通常會(huì)出現(xiàn)這種情況。

      中間件和監(jiān)聽

      Middleware And Listeners

      中間件

      Middleware

      有兩種類型的中間件: 請求和響應(yīng)。兩者都是使用@app聲明的。中間件裝飾器,裝飾器的參數(shù)是一個(gè)代表其類型的字符串:“請求”或“響應(yīng)”。響應(yīng)中間件接收請求和響應(yīng)作為參數(shù)。

      最簡單的中間件根本不修改請求或響應(yīng)

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

      修改請求或響應(yīng)

      中間件可以修改它所提供的請求或響應(yīng)參數(shù),只要它不返回它

      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)

      上述代碼將按順序應(yīng)用這兩個(gè)中間件。首先,中間件custombanner將把HTTP響應(yīng)頭服務(wù)器更改為假服務(wù)器,而第二個(gè)中間件防止XSS將添加HTTP頭來防止跨站點(diǎn)腳本攻擊(XSS)攻擊。這兩個(gè)函數(shù)是在用戶函數(shù)返回響應(yīng)之后調(diào)用的。

      監(jiān)聽者

      Listeners

      如果想在服務(wù)器啟動(dòng)或關(guān)閉時(shí)執(zhí)行啟動(dòng)/分解代碼,可以使用以下偵聽器:

      • before_server_start
      • after_server_start
      • before_server_stop
      • after_server_stop

      這些監(jiān)聽器在函數(shù)中實(shí)現(xiàn)為修飾符,它們接受應(yīng)用程序?qū)ο蠛蚢syncio循環(huán)

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

      如果你想在循環(huán)開始后運(yùn)行一個(gè)后臺(tái)任務(wù),那么Sanic就提供了addtask方法來輕松地完成這一任務(wù)。

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

      藍(lán)圖

      Blueprints

      藍(lán)圖是可以用于應(yīng)用程序中的子路由的對象。除了向應(yīng)用程序?qū)嵗砑勇酚?,藍(lán)圖還定義了類似的添加路由的方法,然后以靈活的可插入的方式在應(yīng)用程序中注冊。

      simple Blueprint

      假設(shè)將該文件保存為myblueprint。py,稍后可以導(dǎo)入到您的主應(yīng)用程序中。

      from sanic.response import json
      from sanic import Blueprint
      
      bp = Blueprint('my_blueprint')
      
      @bp.route('/')
      async def bp_root(request):
          return json({'my': 'blueprint'})

      注冊藍(lán)圖

      registering blueprints

      藍(lán)圖必須在應(yīng)用程序中注冊

      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)

      使用藍(lán)圖

      Use_blueprint

      網(wǎng)絡(luò)套接字路由

      WebSocket routes

      WebSocket處理程序可以注冊,使用@bp.websocket裝飾或bp.add_websocket_route方法

      中間件

      Middleware

      使用藍(lán)圖還可以在全局內(nèi)注冊中間件。

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

      異常

      Exception

      異常情況可以用于全局的藍(lán)圖

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

      靜態(tài)文件

      Static files

      靜態(tài)文件可以添加前綴

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

      Start and stop

      藍(lán)圖可以在服務(wù)器的啟動(dòng)和停止過程中運(yùn)行函數(shù)。如果在多處理器模式下運(yùn)行(超過1個(gè)worker),這些都是在workers fork之后觸發(fā)的。

      • before_server_start:在服務(wù)器開始接受連接之前執(zhí)行
      • after_server_start:在服務(wù)器開始接受連接后執(zhí)行
      • before_server_stop:在服務(wù)器停止接受連接之前執(zhí)行
      • after_server_stop:在服務(wù)器停止后執(zhí)行,所有請求都完成了
      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()

      用例:API版本控制

      Use-case: API versioning

      藍(lán)圖對于API版本控制是非常有用的,其中一個(gè)藍(lán)圖可能指向/v1/<route>,另一個(gè)指向/v2/<route>。

      當(dāng)一個(gè)藍(lán)圖被初始化時(shí),它可以選擇一個(gè)可選的url_prefix參數(shù),這個(gè)參數(shù)將被預(yù)先定義到藍(lán)圖中定義的所有路由。該特性可用于實(shí)現(xiàn)我們的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')

      當(dāng)我們在應(yīng)用程序上注冊我們的藍(lán)圖時(shí),路徑/v1和/v2將指向單個(gè)的藍(lán)圖,它允許為每個(gè)API版本創(chuàng)建子站點(diǎn)。

      # 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)

      用url_for構(gòu)建url

      如果希望在blueprint內(nèi)部路由生成一個(gè)URL,記住,端點(diǎn)名稱采用格式<blueprint_name>.<handler_name>

      @blueprint_v1.route('/')
      async def root(request):
          url = app.url_for('v1.post_handler', post_id=5) # --> '/v1/post/5'
          # url = request.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))

      Note: 當(dāng)app和blueprint不在同一個(gè)模塊內(nèi)記得加上request

      例如:url = request.app.url_for('v1.post_handler', post_id=5) # --> '/v1/post/5'

      配置

      Configuration

      任何一個(gè)相當(dāng)復(fù)雜的應(yīng)用程序都需要配置,而不是在實(shí)際代碼中進(jìn)行。對于不同的環(huán)境或安裝,設(shè)置可能是不同的。

      基本配置

      Sanic在應(yīng)用程序?qū)ο蟮呐渲脤傩灾斜3峙渲?。配置對象僅僅是一個(gè)可以使用點(diǎn)符號或字典來修改的對象。

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

      因?yàn)榕渲脤ο髮?shí)際上是一個(gè)字典,所以可以使用它的update方法來一次設(shè)置幾個(gè)值:

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

      一般來說,該約定只具有大寫的配置參數(shù)。下面描述的加載配置的方法只會(huì)查找這些大寫的參數(shù)。

      載入配置

      如何加載配置有幾種方法。

      從環(huán)境變量

      使用SANIC_前綴定義的任何變量都將應(yīng)用于sanic config。例如,設(shè)置SANIC_REQUEST_TIMEOUT將由應(yīng)用程序自動(dòng)加載并輸入REQUEST_TIMEOUT配置變量。你可以通過一個(gè)不同的前綴到Sanic:

      app = Sanic(load_env='MYAPP_')

      然后,上面的變量將是MYAPP_REQUEST_TIMEOUT。如果您想要禁用環(huán)境變量的加載,您可以將其設(shè)置為False:

      app = Sanic(load_env=False)

      從對象

      如果有很多配置值,而且它們有合理的默認(rèn)值,那么將它們放到一個(gè)模塊中可能會(huì)有幫助:

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

      您也可以使用類或任何其他對象。

      從文件

      通常,您需要從一個(gè)不屬于分布式應(yīng)用程序的文件中加載配置。可以使用from_pyfile(/path/to/config_file)從文件加載配置。但是,這需要程序知道配置文件的路徑。因此,您可以在一個(gè)環(huán)境變量中指定配置文件的位置,并告訴Sanic使用它來查找配置文件:

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

      然后,您可以使用MYAPP_SETTINGS環(huán)境變量集運(yùn)行您的應(yīng)用程序:

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

      配置文件是常規(guī)的Python文件,這些文件是為了加載它們而執(zhí)行的。這允許您使用任意邏輯來構(gòu)造正確的配置。在配置中只添加了大寫的變量。最常見的配置包括簡單的鍵值對:

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

      內(nèi)置配置值

      在這個(gè)框中,只有幾個(gè)預(yù)定義值,在創(chuàng)建應(yīng)用程序時(shí)可以重寫。

      | Variable | Default | Description | | ------------------ | --------- | --------------------------------------------- | | REQUEST_MAX_SIZE | 100000000 | How big a request may be (bytes) | | REQUEST_TIMEOUT | 60 | How long a request can take to arrive (sec) | | RESPONSE_TIMEOUT | 60 | How long a response can take to process (sec) | | KEEP_ALIVE | True | Disables keep-alive when False | | KEEP_ALIVE_TIMEOUT | 5 | How long to hold a TCP connection open (sec) |

      不同的超時(shí)變量

      請求超時(shí)度量在新打開的TCP連接被傳遞給Sanic后端服務(wù)器時(shí)的時(shí)間間隔,以及接收整個(gè)HTTP請求的瞬間。如果時(shí)間超過了REQUEST_TIMEOUT值(以秒為單位),那么這將被視為客戶端錯(cuò)誤,因此Sanic生成一個(gè)HTTP 408響應(yīng)并將其發(fā)送給客戶端。如果您的客戶經(jīng)常通過非常大的請求負(fù)載或者非常緩慢地上傳請求,請調(diào)整這個(gè)值。

      響應(yīng)超時(shí)度量在Sanic服務(wù)器將HTTP請求傳遞到Sanic應(yīng)用程序的時(shí)間之間的時(shí)間,以及發(fā)送到客戶機(jī)的HTTP響應(yīng)的時(shí)間。如果時(shí)間超過了RESPONSE_TIMEOUT值(以秒為單位),這被認(rèn)為是服務(wù)器錯(cuò)誤,因此Sanic生成一個(gè)HTTP 503響應(yīng)并將其設(shè)置為客戶機(jī)。如果應(yīng)用程序很可能長時(shí)間運(yùn)行,延遲響應(yīng)的生成,則將此值調(diào)整得更高。

      Keep-Alive是什么? Keep Alive Timeout value的作用是什么呢?

      Keep-Alive是HTTP 1.1中的一個(gè)HTTP特性。發(fā)送HTTP請求時(shí),客戶端(通常是web瀏覽器應(yīng)用程序)可以設(shè)置一個(gè)keepalive消息頭,以指示HTTP服務(wù)器(Sanic)在發(fā)送響應(yīng)之后不關(guān)閉TCP連接。這允許客戶端重用現(xiàn)有的TCP連接來發(fā)送后續(xù)的HTTP請求,并確??蛻魴C(jī)和服務(wù)器的網(wǎng)絡(luò)流量更高效。

      在Sanic中,KEEP_ALIVE配置變量默認(rèn)設(shè)置為True。如果您在應(yīng)用程序中不需要此功能,則將其設(shè)置為False,以便在發(fā)送響應(yīng)后立即關(guān)閉所有客戶端連接,而不考慮請求上的keepalive消息頭。

      服務(wù)器保持TCP連接打開的時(shí)間量由服務(wù)器本身決定。在Sanic中,該值使用KEEP_ALIVE_TIMEOUT值進(jìn)行配置。默認(rèn)情況下,它設(shè)置為5秒,這是與Apache HTTP服務(wù)器相同的默認(rèn)設(shè)置,在允許客戶端發(fā)送新請求的足夠時(shí)間和不同時(shí)打開太多連接之間保持良好的平衡。不要超過75秒,除非你知道你的客戶正在使用支持TCP連接的瀏覽器。

      供參考:

      Apache httpd server default keepalive timeout = 5 seconds
      Nginx server default keepalive timeout = 75 seconds
      Nginx performance tuning guidelines uses keepalive = 15 seconds
      IE (5-9) client hard keepalive limit = 60 seconds
      Firefox client hard keepalive limit = 115 seconds
      Opera 11 client hard keepalive limit = 120 seconds
      Chrome 13+ client keepalive limit > 300+ seconds

      Cookie

      cookie是保存在用戶瀏覽器內(nèi)的數(shù)據(jù)塊。Sanic既可以讀寫cookie,也可以存儲(chǔ)為鍵-值對。

      Warning

      cookie可以由客戶機(jī)自由修改。因此,您不能僅在cookie中存儲(chǔ)諸如登錄信息這樣的數(shù)據(jù),因?yàn)榭蛻魴C(jī)可以隨意更改這些數(shù)據(jù)。為了確保存儲(chǔ)在cookie中的數(shù)據(jù)不會(huì)被客戶偽造或篡改, use something like itsdangerous to cryptographically sign the data.

      讀取cookies

      用戶的cookie可以通過請求對象的cookie字典訪問。

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

      寫入cookies

      返回響應(yīng)時(shí),可以在響應(yīng)對象上設(shè)置cookie。

      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

      刪除cookies

      cookie可以通過語義或顯式刪除。

      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
      響應(yīng)cookie可以設(shè)置為字典值,并具有以下參數(shù):
      • expires (datetime): cookie在客戶機(jī)瀏覽器上過期的時(shí)間。
      • path(string): 此cookie應(yīng)用的url的子集。默認(rèn)為/。
      • comment(string): 注釋(元數(shù)據(jù))。
      • domain(string): 指定cookie有效的域。顯式指定的域必須總是以一個(gè)點(diǎn)開始。
      • max-age(number): cookie應(yīng)該存活的秒數(shù)。
      • secure (boolean): 指定cookie是否只通過HTTPS發(fā)送。
      • httponly (boolean): 指定Javascript是否不能讀取cookie。

      session

      sanic對此有一個(gè)第三方插件sanic_session,用法非常簡單,見官方例子如下:

      import asyncio_redis
      
      from sanic import Sanic
      from sanic.response import text
      from sanic_session import RedisSessionInterface
      
      app = Sanic()
      
      
      # Token from https://github.com/subyraman/sanic_session
      
      class Redis:
          '''
          A simple wrapper class that allows you to share a connection
          pool across your application.
          '''
          _pool = None
      
          async def get_redis_pool(self):
              if not self._pool:
                  self._pool = await asyncio_redis.Pool.create(
                      host='localhost', port=6379, poolsize=10
                  )
      
              return self._pool
      
      
      redis = Redis()
      
      # pass the getter method for the connection pool into the session
      session_interface = RedisSessionInterface(redis.get_redis_pool, expiry=604800)
      
      
      @app.middleware('request')
      async def add_session_to_request(request):
          # before each request initialize a session
          # using the client's request
          await session_interface.open(request)
      
      
      @app.middleware('response')
      async def save_session(request, response):
          # after each request save the session,
          # pass the response to set client cookies
          await session_interface.save(request, response)
      
      
      @app.route('/')
      async def test(request):
          # interact with the session like a normal dict
          if not request['session'].get('foo'):
              request['session']['foo'] = 0
      
          request['session']['foo'] += 1
      
          response = text(request['session']['foo'])
      
          return response
      
      
      if __name__ == '__main__':
          app.run(host='0.0.0.0', port=8888, debug=True)

      Handler Decorators

      由于Sanic處理程序是簡單的Python函數(shù),您可以用類似于Flask的方式向它們應(yīng)用decorator。一個(gè)典型的用例是,當(dāng)您需要一些代碼在處理程序的代碼執(zhí)行之前運(yùn)行。

      Authorization Decorator

      假設(shè)您想要檢查用戶是否被授權(quán)訪問某個(gè)特定的端點(diǎn)。您可以創(chuàng)建包裝處理函數(shù)的decorator,檢查客戶端是否被授權(quán)訪問某個(gè)資源,并發(fā)送適當(dāng)?shù)捻憫?yīng)。

      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'})

      Streaming

      流媒體

      Request Streaming

      Sanic允許您通過流獲取請求數(shù)據(jù),如下所示。當(dāng)請求結(jié)束時(shí),request.stream.get()返回None。只有post, put和patch decorator 有流參數(shù)。

      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)

      Response Streaming

      Sanic允許您使用stream 方法將內(nèi)容流到客戶機(jī)。該方法接受一個(gè)coroutine回調(diào),該回調(diào)將傳遞給寫入的StreamingHTTPResponse對象。一個(gè)簡單的例子如下:

      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')

      在您希望將內(nèi)容流到來自外部服務(wù)(如數(shù)據(jù)庫)的客戶端時(shí),這很有用。例如,您可以使用asyncpg提供的異步游標(biāo)將數(shù)據(jù)庫記錄流到客戶端:

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

      基于類的視圖

      基于類的視圖只是實(shí)現(xiàn)對請求的響應(yīng)行為的類。它們提供了一種方法,將不同的HTTP請求類型劃分到同一端點(diǎn)。與其定義和修飾三個(gè)不同的處理函數(shù)(每個(gè)端點(diǎn)支持的請求類型),端點(diǎn)可以分配一個(gè)基于類的視圖。

      定義視圖

      基于類的視圖應(yīng)該子類化HTTPMethodView。然后,您可以為希望支持的每個(gè)HTTP請求類型實(shí)現(xiàn)類方法。如果接收到的請求沒有定義的方法,則會(huì)生成一個(gè)405: Method not allowed 的響應(yīng)。

      要在端點(diǎn)上注冊基于類的視圖,將使用app.add_route方法。第一個(gè)參數(shù)應(yīng)該是被調(diào)用的方法as_view的定義類,第二個(gè)參數(shù)應(yīng)該是URL端點(diǎn)。

      可用的方法是get、postputpatchdelete。使用所有這些方法的類看起來如下所示。

      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(), '/')

      URL參數(shù)

      如果您需要任何URL參數(shù),如路由指南中所討論的,在方法定義中包含它們。

      class NameView(HTTPMethodView): def get(self, request, name): return text('Hello {}'.format(name)) app.add_route(NameView.as_view(), '/<name>')

      裝飾器

      如果您想在類中添加任何裝飾器,可以設(shè)置decorator類變量。當(dāng)調(diào)用as_view時(shí),這些將被應(yīng)用到類中。

      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')

      URL構(gòu)建

      如果您希望為HTTPMethodView構(gòu)建一個(gè)URL,請記住,類名將是您將傳入url_for的端點(diǎn)。例如:

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

      使用組合視圖

      Using CompositionView

      作為HTTPMethodView的替代方法,您可以使用CompositionView將處理程序函數(shù)移到視圖類之外。

      每個(gè)支持的HTTP方法的處理函數(shù)都在源代碼的其他地方定義,然后使用CompositionView.add方法添加到視圖中。第一個(gè)參數(shù)是要處理的HTTP方法的列表(例如,['GET', 'POST']),第二個(gè)參數(shù)是處理函數(shù)。下面的示例顯示了使用外部處理程序函數(shù)和內(nèi)聯(lián)lambda的CompositionView用法:

      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, '/')

      Note: 當(dāng)前您不能使用url_for為CompositionView構(gòu)建一個(gè)URL。

      自定義協(xié)議

      注意:這是高級用法,大多數(shù)讀者不需要這樣的功能。

      您可以通過指定自定義協(xié)議來更改Sanic協(xié)議的行為,該協(xié)議應(yīng)該是asyncio.protocol的子類。然后,該協(xié)議可以作為sanic.run方法的關(guān)鍵字參數(shù)協(xié)議傳遞。

      自定義協(xié)議類的構(gòu)造函數(shù)接收來自Sanic的以下關(guān)鍵字參數(shù)。

      • loop: 一個(gè)異步兼容的事件循環(huán)。
      • connections: 用于存儲(chǔ)協(xié)議對象的集合。當(dāng)Sanic接收SIGINTSIGTERM時(shí),它執(zhí)行protocol.close_if_idle關(guān)閉此集合中存儲(chǔ)的所有協(xié)議對象。
      • signal: 帶有stopped屬性的sanic.server.Signal對象。當(dāng)Sanic收到SIGINTSIGTERM,signal.stopped分配True。
      • request_handler: 取一個(gè)sanic.request.Request對象和response回調(diào)作為參數(shù)的coroutine。
      • error_handler: 在拋出異常時(shí)調(diào)用的處理程序sanic.exceptions.Handler。
      • request_timeout: 請求超時(shí)前的秒數(shù)。
      • request_max_size: 指定請求的最大大小的整數(shù),以字節(jié)為單位。

      Example

      如果處理函數(shù)不返回HTTPResponse對象,則默認(rèn)協(xié)議中出現(xiàn)錯(cuò)誤。

      通過重寫write_response協(xié)議方法,如果處理程序返回一個(gè)字符串,它將被轉(zhuǎn)換為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)

      SSL Example

      可以傳入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)

      日志

      Logging

      Sanic允許您根據(jù)python3 logging API對請求進(jìn)行不同類型的日志記錄(訪問日志、錯(cuò)誤日志)。如果您想創(chuàng)建一個(gè)新的配置,您應(yīng)該對python3 logging有一些基本的了解。

      Quick Start

      使用默認(rèn)設(shè)置的一個(gè)簡單示例如下:

      from sanic import Sanic
      
      app = Sanic('test')
      
      @app.route('/')
      async def test(request):
          return response.text('Hello World!')
      
      if __name__ == '__main__':
        app.run(debug=True, access_log=True)

      要使用自己的日志記錄配置,只需使用logging.config.dictConfig,或在初始化Sanic應(yīng)用時(shí)傳遞log_config:

      app = Sanic('test', log_config=LOGGING_CONFIG)

      要關(guān)閉日志,只需分配access_log=False:

      if __name__ == '__main__':
        app.run(access_log=False)

      這將跳過在處理請求時(shí)調(diào)用日志功能。你甚至可以做進(jìn)一步的生產(chǎn)以獲得額外的速度:

      if __name__ == '__main__': # disable debug messages app.run(debug=False, access_log=False)

      Configuration

      默認(rèn)情況下,log_config參數(shù)設(shè)置為使用sanic.log.LOGGING_CONFIG_DEFAULTS字典配置。

      在sanic中使用了三個(gè)日志記錄器loggers,如果您想創(chuàng)建自己的日志配置,則必須定義:

      root: 用于記錄內(nèi)部消息。

      sanic.error: 用于記錄錯(cuò)誤日志。

      sanic.access: 用于記錄訪問日志。

      Log format

      除了python提供的默認(rèn)參數(shù)(asctime、levelname、message), Sanic還提供了用于訪問日志記錄器logger的其他參數(shù):

      host (str): request.ip
      request (str): request.method + ' ' + request.url
      status (int): response.status
      byte (int): len(response.body)

      默認(rèn)的訪問日志格式是:

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

      測試

      Testing

      Sanic端點(diǎn)可以使用test_client對象在本地測試,這取決于附加的aiohttp庫。

      test_client公開getpost、putdelete、patch、headoptions方法,以便與應(yīng)用程序運(yùn)行。一個(gè)簡單的例子(使用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

      在內(nèi)部,每次調(diào)用test_client方法時(shí),Sanic應(yīng)用程序運(yùn)行于127.0.0.1:42101,您的測試請求使用aiohttp執(zhí)行。

      test_client方法接受以下參數(shù)和關(guān)鍵字參數(shù):

      uri(default '/') 一個(gè)表示測試uri的字符串。

      gather_request(default True) 一個(gè)布爾值,它決定原始請求是否由函數(shù)返回。如果設(shè)置為True,返回值是(request, response)的一個(gè)元組,如果False僅返回響應(yīng)。

      server_kwargs (default {}) 在測試請求運(yùn)行之前傳遞給app.run的附加參數(shù)。

      debug(default False)一個(gè)布爾值,它決定是否在調(diào)試模式下運(yùn)行服務(wù)器。

      該函數(shù)進(jìn)一步接受了*request_args和**request_kwargs,它們直接傳遞給aiohttp ClientSession請求。

      例如,為了向GET請求提供數(shù)據(jù),您將執(zhí)行以下操作:

      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請求提供數(shù)據(jù):

      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'

      關(guān)于aiohttp的可用參數(shù)的更多信息可以在ClientSession的文檔中找到。

      pytest-sanic

      pytest-sanic是一個(gè)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還提供了一些有用的設(shè)備,如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}

      部署

      Deploying

      部署Sanic是由內(nèi)置的webserver簡化的。在定義了sanic.Sanic的實(shí)例之后。我們可以用以下關(guān)鍵字參數(shù)調(diào)用run方法:

      • host (default '127.0.0.1'): 地址來托管服務(wù)器。
      • port (default 8000): 開啟服務(wù)器的端口。
      • debug (default False): 啟用調(diào)試輸出(減慢服務(wù)器)。
      • ssl (default None): ssl加密的SSLContext。
      • sock (default None): 用于服務(wù)器接受連接的套接字。
      • workers (default 1): 生成的工作進(jìn)程數(shù)。
      • loop (default None): 一個(gè)asyncio兼容的事件循環(huán)。如果沒有指定,Sanic將創(chuàng)建自己的事件循環(huán)。
      • protocol (default HttpProtocol): asyncio.protocol的子類。

      進(jìn)程

      Workers

      默認(rèn)情況下,Sanic只使用一個(gè)CPU核心偵聽主進(jìn)程。To crank up the juice,只需在run參數(shù)中指定workers的數(shù)量。

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

      Sanic將會(huì)自動(dòng)啟動(dòng)多個(gè)進(jìn)程,并在它們之間路由流量。我們建議盡可能多的workers擁有可用的核心。

      通過命令行運(yùn)行

      如果您喜歡使用命令行參數(shù),則可以通過執(zhí)行模塊來啟動(dòng)Sanic服務(wù)器。例如,如果您在名為server.py的文件中初始化Sanic作為app,你可以這樣運(yùn)行服務(wù)器:

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

      通過這種運(yùn)行sanic的方式,無需在Python文件中調(diào)用app.run。如果這樣做,請確保將其包裝起來,以便它只在解釋器直接運(yùn)行時(shí)執(zhí)行。

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

      通過Gunicorn運(yùn)行

      Gunicorn 'Green Unicorn’ 是UNIX的一個(gè)WSGI HTTP服務(wù)器。它是一個(gè)由Ruby的Unicorn項(xiàng)目移植的預(yù)fork worker模型。

      為了使用Gunicorn運(yùn)行Sanic應(yīng)用程序,您需要為Gunicornworker-class 參數(shù)使用特殊的sanic.worker.GunicornWorker

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

      如果您的應(yīng)用程序遭受內(nèi)存泄漏,您可以配置Gunicorn以優(yōu)雅地重新啟動(dòng)一個(gè)worker,因?yàn)樗幚砹私o定數(shù)量的請求。這可以方便地幫助限制內(nèi)存泄漏的影響。

      有關(guān)更多信息,請參見 Gunicorn 文檔。

      異步支持

      Asynchronous support

      如果您需要與其他應(yīng)用程序共享sanic進(jìn)程,特別是loop,這是合適的。但是,請注意,該方法不支持使用多進(jìn)程,并且不是一般運(yùn)行該應(yīng)用程序的首選方式。

      這里有一個(gè)不完整的示例(請參見run_asyn.py在一些更實(shí)用的例子中):

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

      擴(kuò)展

      Extensions

      由社區(qū)創(chuàng)建的Sanic擴(kuò)展列表。

      • Sanic-Plugins-Framework: 方便創(chuàng)建和使用Sanic插件的庫。
      • Sessions: 對sessions的支持。允許使用redis、memcache或內(nèi)存存儲(chǔ)。
      • CORS: A port of flask-cors.
      • Compress: 允許您輕松地壓縮Sanic響應(yīng)。Flask-Compress的一個(gè)端口。
      • Jinja2: 支 持Jinja2模板。
      • JWT: JSON Web令牌(JWT)的身份驗(yàn)證擴(kuò)展。
      • OpenAPI/Swagger: OpenAPI支持,外加Swagger UI。
      • Pagination: 簡單的分頁的支持。
      • Motor: Simple motor wrapper。
      • Sanic CRUD: 與peewee模型的CRUD REST API生成。
      • UserAgent: Add user_agent to request。
      • Limiter: sanic的速率限制。
      • Sanic EnvConfig: 將環(huán)境變量引入到Sanic配置中。
      • Babel: 在Babel庫的幫助下向Sanic應(yīng)用程序添加i18n/l10n支持。
      • Dispatch: 在werkzeug中由DispatcherMiddleware激發(fā)的調(diào)度程序??梢猿洚?dāng)sanicto - wsgi適配器。
      • Sanic-OAuth: OAuth庫,用于連接和創(chuàng)建您自己的令牌提供者。
      • Sanic-nginx-docker-example: 使用Sanic構(gòu)建簡單易用的骨架項(xiàng)目需要使用nginx,由docker-compose編排。
      • sanic-graphql: Graphico與Sanic集成。
      • sanic-prometheus: Sanic的Prometheus標(biāo)準(zhǔn)。
      • Sanic-RestPlus: A port of Flask-RestPlus for Sanic. Full-featured REST API with SwaggerUI generation。
      • sanic-transmute: 從python函數(shù)和類中生成api的Sanic擴(kuò)展,并自動(dòng)生成Swagger UI文檔。
      • pytest-sanic: Sanic的一個(gè)pytest插件。它幫助您異步地測試代碼。
      • jinja2-sanic: Sanic的jinja2模板渲染器。

      API Reference

      sanic-官方文檔

      sanic-githup

        本站是提供個(gè)人知識(shí)管理的網(wǎng)絡(luò)存儲(chǔ)空間,所有內(nèi)容均由用戶發(fā)布,不代表本站觀點(diǎn)。請注意甄別內(nèi)容中的聯(lián)系方式、誘導(dǎo)購買等信息,謹(jǐn)防詐騙。如發(fā)現(xiàn)有害或侵權(quán)內(nèi)容,請點(diǎn)擊一鍵舉報(bào)。
        轉(zhuǎn)藏 分享 獻(xiàn)花(0

        0條評論

        發(fā)表

        請遵守用戶 評論公約

        類似文章 更多