aiohttp-security/docs/example.rst

70 lines
2.3 KiB
ReStructuredText
Raw Normal View History

.. _aiohttp-security-example:
===============================================
2015-07-08 17:30:24 +00:00
How to Make a Simple Server With Authorization
===============================================
2015-07-08 17:30:24 +00:00
Simple example::
2015-07-08 17:30:24 +00:00
import asyncio
from aiohttp import web
2017-12-13 14:51:46 +00:00
async def root_handler(request):
2015-07-08 17:30:24 +00:00
text = "Alive and kicking!"
return web.Response(body=text.encode('utf-8'))
# option 2: auth at a higher level?
# set user_id and allowed in the wsgi handler
2015-07-08 17:30:24 +00:00
@protect('view_user')
2017-12-13 14:51:46 +00:00
async def user_handler(request):
2015-07-08 17:30:24 +00:00
name = request.match_info.get('name', "Anonymous")
text = "Hello, " + name
return web.Response(body=text.encode('utf-8'))
# option 3: super low
# wsgi doesn't do anything
2017-12-13 14:51:46 +00:00
async def user_update_handler(request):
2015-07-08 17:30:24 +00:00
# identity, asked_permission
2017-12-13 14:51:46 +00:00
user_id = await identity_policy.identify(request)
identity = await auth_policy.authorized_userid(user_id)
allowed = await request.auth_policy.permits(
identity, asked_permission)
2015-07-08 17:30:24 +00:00
if not allowed:
# how is this pluggable as well?
# ? return NotAllowedStream()
raise NotAllowedResponse()
update_user()
2017-12-13 14:51:46 +00:00
async def init(loop):
2015-07-08 17:30:24 +00:00
# set up identity and auth
auth_policy = DictionaryAuthorizationPolicy({'me': ('view_user',),
2015-11-26 12:09:01 +00:00
'you': ('view_user',
2015-07-08 17:30:24 +00:00
'edit_user',)})
identity_policy = CookieIdentityPolicy()
auth = authorization_middleware(auth_policy, identity_policy)
# wsgi app
app = web.Application(loop=loop, middlewares=*auth)
# add the routes
app.router.add_route('GET', '/', root_handler)
app.router.add_route('GET', '/{user}', user_handler)
app.router.add_route('GET', '/{user}/edit', user_update_handler)
# get it started
2017-12-13 14:51:46 +00:00
srv = await loop.create_server(app.make_handler(),
'127.0.0.1', 8080)
2015-07-08 17:30:24 +00:00
print("Server started at http://127.0.0.1:8080")
return srv
loop = asyncio.get_event_loop()
loop.run_until_complete(init(loop))
try:
loop.run_forever()
except KeyboardInterrupt:
pass # TODO put handler cleanup here