1,websocket(网络套接字)
在 websocket 上 Sanic 提供了一种简单使用的抽象化, 来设置 websocket(网络套接字)
- from sanic import Sanic
- from sanic.response import JSON
- from sanic.websocket import WebSocketCommonProtocol
- # 实例化一个 Sanic 对象
- App = Sanic()
- @App.websocket("/feed")
- async def feed(request, ws):
- while 1:
- data = "hello!"
- print("Sending:" + data)
- # 阻塞发送数据
- await ws.send(data)
- # 阻塞接收数据
- data = await ws.recv()
- print("Received:" + data)
- if __name__ == '__main__':
- App.run(host="0.0.0.0", port=8000, protocol=WebSocketCommonProtocol) # 指定协议是 WebsocketCommonProtocol
App.add_websocket_route 方法能够被替换成路由的装饰器
- from sanic import Sanic
- App = Sanic()
- async def feed(request, ws):
- pass
- App.add_websocket_route(feed, "/feed")
调用 WebSocket 路由的处理程序时, 请求作为第一个参数, WebSocket 协议对象作为第二个参数. 协议对象具有分别发送和接收数据的发送和接收方法
你能够通过 App.config 来设置自己的 Websocket 配置.
- App.config.WEBSOCKET_MAX_SIZE = 2 ** 20
- App.config.WEBSOCKET_MAX_QUEUE = 32
- App.config.WEBSOCKET_READ_LIMIT = 2 **16
- App.config.WEBSOCKET_WRITE_LIMIT = 2 ** 16
2, 处理程序的装饰器
自从 Sanic 处理程序时简单的 Python 函数 (功能), 你可以使用装饰器去装饰函数在单一的管理在 flask 中, 一种典型的使用, 一个典型的用例就是, 当你希望执行处理代码之前, 运行一些代码时候
2.1 授权装饰器
假设您希望检查用户是否有权访问特定端点. 您可以创建一个包装处理程序函数的装饰器, 检查客户机是否有权访问资源, 并发送适当的响应.
- from functools import wraps
- from sanic.response import JSON
- from sanic import Sanic
- # 实例化 sanic 对象
- App = Sanic()
- def authorized():
- def decorator(f):
- @wraps(f)
- async def decorated_function(request, *args, **kwargs):
- # 检查运行方法中的 request 请求
- # 客户授权状态
- is_authorized = check_request_for_authorization_status(request) # 执行查看请求校验状态的函数
- # is_authorized 条件成立
- if is_authorized:
- # 说明这个用户已经已经被授权了
- # 运行这个处理程序的方法并且返回响应
- response = await f(request, *args, **kwargs)
- return response
- else:
- # 否则这个用户没有被授权
- # 返回 JSON 格式的数据
- return JSON({"status": "not_authorized"}, 403)
- return decorated_function
- return decorator
- @App.route("/") # 访问路由
- @authorized() # 授权的验证
- async def test(request): # 被装饰的函数
- return JSON({"status": authorized})
来源: http://www.bubuko.com/infodetail-2958116.html