Sanic 基礎應用 -1

Sanic 基礎 - 1

官方文檔

1.啓 動sanic

1. 安裝:

   pip install sanic

2. 啓動

   from sanic import Sanic
   from sanic.response import text, json
   
   app = Sanic()
   
   # 必須使用async def語法定義函數，來保證其能夠進行異步處理。
   @app.route('/')  # 默認 GET 請求
   async def index(request):
       return json({"code": 0, "data": "請求成功"})
   
   
   if __name__ == '__main__':
       app.run(host='0.0.0.0', port=5000, workers=4) # 指定使用4個cpu

3. run 參數說明

   host(默認「127.0.0.1」): 服務器主機的地址。
   port(默認8000): 服務器的端口。
   debug(默認False): 啓用調試（減慢服務器速度）。
   ssl(默認None): 用於工做者SSL加密的SSLContext。
   sock（默認None）：服務器接受鏈接的Socket。
   worker（默認值1）：生成的工做進程數。
   loop（默認None）: asyncio兼容的事件循環。若是沒有指定，Sanic會建立本身的事件循環。
   protocol（默認HttpProtocol）：asyncio.protocol的子類。

2.路由系統

2.0 路由參數

@app.route("/index",
           strict_slashes=True,  # 嚴格模式 url 沒有最後的 " / "
           name="my_index",  # 路由別名  app.url_for("my_index")
           
          )
async def index(request, tag):

2.1 請求參數 及請求類型

from sanic import Sanic
from sanic.response import json, text

app = Sanic()

# 接受任意類型
@app.route('/index/<tag>', methods=['GET'])
async def index(request, tag):
    print(tag)
    # 參數:  ensure_ascii=False 使用中文  indent=4 使用四格縮進
    res = json({"code": 1, "data": {1: int}}, status=200, ensure_ascii=False, indent=4)
    return res


# 只接受整數  
@app.route('/tag/<int:int>', methods=['GET'])
async def index(request, int):
    print(int)
    # 參數:  ensure_ascii=False 使用中文  indent=4 使用四格縮進
    res = json({"code": 1, "data": {1: int}}, status=200, ensure_ascii=False, indent=4)
    return res


# 接受 浮點數
@app.route('/number/<number:number>')
async def number(request, number):
    print(request, number)
    return text("number--{}".format(number))


# 只接受一個字符
@app.route('/name/<name:[A-z]>')
async def name(request, name):
    print(name)
    # 返回文本類型
    return text("{}".format(name))


# 能夠使用正則  只接受 POST 請求
@app.route('/re/<re:[\w]{0,4}>', methods=['POST'])
async def folder_handler(request, re):
    # 返回文本類型
    return text('re - {}'.format(re))

2.2 直接指定請求類型

@app.get('/get')
async def get(request):
    return text(f"{request.args}")


@app.post('/post')
async def get(request):
    return text(f"{request.json}")

2.3 方法添加路由

async def demo1(request):
    return one('OK')


async def demo2(request, name):
    return tow('Folder - {}'.format(name))


async def demo3(request, name):
    return three('Person - {}'.format(name))


app.add_route(demo1, '/one')
app.add_route(demo2, '/tow/<name>')
app.add_route(demo3, '/three/<name:[A-z]>', methods=['GET'])

2.4 CompositionView 添加路由

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

app = Sanic(__name__)

async def post_handler(request):
    print(request.json)
    return text('it is ok!')


view = CompositionView()
view.add(["POST"], post_handler)
app.add_route(view, "/post_info")

2.5 可根請求的域名 來區分返回內容

# ---------------------------  可根請求的域名 來區分返回內容  -------------------------

from sanic import Sanic
from sanic.response import text

app = Sanic()


@app.route('/get', methods=['GET'], host='zhangfei.com:5000')
async def get_handler(request):
    return text('GET request - {}'.format(request.args))


# 若是主機頭與zhangfei.com不匹配，將使用此路由
@app.route('/get', methods=['GET'])
async def get_handler(request):
    return text('GET request in default - {}'.format(request.args))

2.6 利用url_for生成URL

from sanic import Sanic
from sanic.response import text, json, redirect

app = Sanic()


@app.route('/')
async def index(request):
    url = app.url_for('demo', tag=10, args=[1, 2, 3], arg_1="aaa")
    return redirect(url)  # 302 狀態碼


@app.route('/demo/<tag>')
async def demo(request, tag):
    # 數字會轉爲字符串  都已列表方式傳遞
    print(request.args)  # {'args': ['1', '2', '3'], 'arg_1': ['aaa']}
    return text('i am demo {}'.format(tag))

2.7 靜態文件 以及 html 頁面返回

from sanic import Sanic, Blueprint

app = Sanic()
app.static("/home", "./static/demo.html")
app.static("/static", "./static")
if __name__ == '__main__':

    app.run()

3. WebSocket 路由系統

from sanic import Sanic
from sanic.response import text, json, redirect

app = Sanic()


@app.websocket('/ws')
async def ws(request, ws):
    while 1:
        data = "鏈接成功!"
        await ws.send('msg = {}'.format(data))
        res = await ws.recv()
        print(res)


async def wss(request, ws):
    pass

# 方法添加路由
app.add_websocket_route(wss, '/wss')

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

• 前端連接

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
<script>
    var ws = new WebSocket('ws://127.0.0.1:5000/ws');
    ws.onmessage = function (res) {
        console.log(res.data)
    }

</script>
</body>
</html>

4. request 對象參數

4.1 request的方法

@app.route('/query', methods=['POST'])
async def ret_query(request):
    print(request.json)  # 接受的 json 數據
    print(request.args)  # {'page': ['1']}
    print(request.url)  # http://127.0.0.1:5000/query?page=1
    print(request.query_string)  # page=1
    print(request.form)  # form 表單提交的數據 {'name': ['123456789']}
    print(request.body)  # 這個屬性容許檢索請求的原始數據  b' {"desk_id": 44}'
    print(request.ip)  # 請求者的IP地址
    print(request.scheme)  # 請求協議  http
    print(request.host)  # 127.0.0.1:5000
    print(request.path)  # 請求者的 路由 /query

    return json({"parsed": True, "args": request.args, "url": request.url, "query_string": request.query_string})

4.2接收文件

@app.route('/files', methods=['POST'])
async def ret_file(request):
    file = request.files.get('filename')
    print(request.body)  # 這個屬性容許檢索請求的原始數據
    print(request.form)  # form 表單提交的數據 {'name': ['123456789']}
    print(file.type)  # image/jpeg
    print(file.name)  # aaa.jpg
    with open(file.name, 'wb') as f:
        f.write(file.body)
    return text('ok')

4.3 app 配置模式

# app.config['DEBUG'] = True
app.config['DEBUG'] = False


@app.route('/json', methods=['POST'])
async def ret_json(request):
    if request.app.config['DEBUG']:
        return json({'status': 'debug'})
    else:
        return json({'status': 'production'})

4.4 原始對象 RequestParameters

from sanic.request import RequestParameters

args = RequestParameters()
args['titles'] = ['Post 1', 'Post 2']

args.get('titles')  # => 'Post 1'

print(args.getlist('titles'))  # => ['Post 1', 'Post 2']

5. response 對象

• text 文本

• html 返回 html 標籤

• file 返回文件

• json 響應 json 數據

• redirect 重定向

• raw 文件會被下載

from sanic import Sanic
from sanic.response import text, json, html, redirect, file, stream, raw

app = Sanic()


@app.route('/')
async def index(request):
    return text('OK')


@app.route('/file')
async def handle_request(request):
    return await file('aaa.jpg')


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


#
@app.route('/raw')
def handle_request(request):
    return raw(b'raw data')


@app.route('/json')
def handle_request(request):
    return json(
        {'message': 'Hello world!'},
        headers={'X-Served-By': 'sanic'},  # 添加響應頭
        status=200
    )


if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)