aiohttp 中基于 Class 的视图
Class-based views in aiohttp
使用基于 class 的处理程序而不是 aiohttp 库中的函数的正确方法是什么?我习惯在 Django 中将处理程序编写为 classes,所以我想知道如何在 aiohttp?
我假设您想使用基于 class 的处理程序来应用继承以重用代码。
从技术上讲,aiohttp 网络处理程序是任何接受请求参数和 returns 响应实例的协程。
例如
class BaseView:
def __init__(self, ...):
...
async def __call__(self, request):
return web.Response()
app.router.add_route('GET', '/', BaseView(...))
可以用作基础 class 来制作网络处理程序层次结构。
甚至
class Handler:
def __init__(self, db):
self._db = db
async def get_from_db(self, data):
...
async def handle_a(self, request):
data = yield from self.get_from_db(
self.extract_from_request_a(request))
return web.Response(self.format_data(data))
async def handle_b(self, request):
data = yield from self.get_from_db(
self.extract_from_request_b(request))
return web.Response(self.format_data(data))
handler = Handler(db)
app.router.add_route('GET', '/a', hadndler.handle_a)
app.router.add_route('GET', '/b', hadndler.handle_b)
你可以这样使用它:
from aiohttp import web
from datetime import datetime
class TokenView(web.View):
async def get(self):
token = datetime.now().strftime("%Y%m%d%H%M%S")
room = self.request.match_info.get("room", None)
return web.json_response({"room": room, "token": token, "result": "OK"})
async def post(self):
room = self.request.match_info.get("room", None)
token = datetime.now().strftime("%Y%m%d%H%M%S")
return web.json_response({"room": room, "token": token, "result": "OK"})
if __name__ == "__main__":
app = web.Application()
app.router.add_view("/token/{room}", TokenView)
print(app.router.named_resources())
web.run_app(app)
在 aiohttp
中基于 Class 的视图 的快速示例
from aiohttp import web
class Users(web.View):
async def get(self):
output = [
{
'id': 1,
'username': 'chuck_norris'
},
]
return web.json_response(output, status=200)
async def post(self):
data = await self.request.json()
output = {
'result': data
}
return web.json_response(output, status=201)
async def delete(self):
return web.json_response(status=204)
class Teams(web.View):
async def get(self):
output = [
{
'id': 1,
'team': 'team1'
},
]
return web.json_response(output, status=200)
async def post(self):
data = await self.request.json()
output = {
'result': data
}
return web.json_response(output, status=201)
async def delete(self):
return web.json_response(status=204)
app = web.Application()
app.router.add_view("/users", Users)
app.router.add_view("/teams", Teams)
web.run_app(app, port=8000)
使用基于 class 的处理程序而不是 aiohttp 库中的函数的正确方法是什么?我习惯在 Django 中将处理程序编写为 classes,所以我想知道如何在 aiohttp?
我假设您想使用基于 class 的处理程序来应用继承以重用代码。
从技术上讲,aiohttp 网络处理程序是任何接受请求参数和 returns 响应实例的协程。
例如
class BaseView:
def __init__(self, ...):
...
async def __call__(self, request):
return web.Response()
app.router.add_route('GET', '/', BaseView(...))
可以用作基础 class 来制作网络处理程序层次结构。
甚至
class Handler:
def __init__(self, db):
self._db = db
async def get_from_db(self, data):
...
async def handle_a(self, request):
data = yield from self.get_from_db(
self.extract_from_request_a(request))
return web.Response(self.format_data(data))
async def handle_b(self, request):
data = yield from self.get_from_db(
self.extract_from_request_b(request))
return web.Response(self.format_data(data))
handler = Handler(db)
app.router.add_route('GET', '/a', hadndler.handle_a)
app.router.add_route('GET', '/b', hadndler.handle_b)
你可以这样使用它:
from aiohttp import web
from datetime import datetime
class TokenView(web.View):
async def get(self):
token = datetime.now().strftime("%Y%m%d%H%M%S")
room = self.request.match_info.get("room", None)
return web.json_response({"room": room, "token": token, "result": "OK"})
async def post(self):
room = self.request.match_info.get("room", None)
token = datetime.now().strftime("%Y%m%d%H%M%S")
return web.json_response({"room": room, "token": token, "result": "OK"})
if __name__ == "__main__":
app = web.Application()
app.router.add_view("/token/{room}", TokenView)
print(app.router.named_resources())
web.run_app(app)
在 aiohttp
中基于 Class 的视图 的快速示例from aiohttp import web
class Users(web.View):
async def get(self):
output = [
{
'id': 1,
'username': 'chuck_norris'
},
]
return web.json_response(output, status=200)
async def post(self):
data = await self.request.json()
output = {
'result': data
}
return web.json_response(output, status=201)
async def delete(self):
return web.json_response(status=204)
class Teams(web.View):
async def get(self):
output = [
{
'id': 1,
'team': 'team1'
},
]
return web.json_response(output, status=200)
async def post(self):
data = await self.request.json()
output = {
'result': data
}
return web.json_response(output, status=201)
async def delete(self):
return web.json_response(status=204)
app = web.Application()
app.router.add_view("/users", Users)
app.router.add_view("/teams", Teams)
web.run_app(app, port=8000)