What is the right approach to use class-based handlers instead of functions in aiohttp library? I'm used to writing handlers as classes in Django so I am wondering how to do it correctly in aiohttp?
You can use it like this:
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)
I assume you want to use class-based handlers for sake of applying inheritance for reusing code.
Technically aiohttp web-handler is any coroutine which accepts request parameter and returns response instance.
For example
class BaseView:
def __init__(self, ...):
...
async def __call__(self, request):
return web.Response()
app.router.add_route('GET', '/', BaseView(...))
can be used as base class for making web-handlers hierarchy.
Or even
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)
Quick example of Class-based views in aiohttp
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)
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With