aiohttp-security/docs/example.rst

.. _aiohttp-security-example:

===============================================
How to Make a Simple Server With Authorization
===============================================


Simple example::

    import asyncio
    from aiohttp import web

    async def root_handler(request):
        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
    @protect('view_user')
    async def user_handler(request):
        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
    async def user_update_handler(request):
        # identity, asked_permission
        user_id = await identity_policy.identify(request)
        identity = await auth_policy.authorized_userid(user_id)
        allowed = await request.auth_policy.permits(
            identity, asked_permission)
        if not allowed:
            # how is this pluggable as well?
            # ? return NotAllowedStream()
            raise NotAllowedResponse()

        update_user()

    async def init(loop):
        # set up identity and auth
        auth_policy = DictionaryAuthorizationPolicy({'me': ('view_user',),
                                                     'you': ('view_user',
                                                             '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
        srv = await loop.create_server(app.make_handler(),
                                       '127.0.0.1', 8080)
        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