aiohttp oauth2

facebook.py

from aiohttp import web
from aiohttp_oauth2.client.contrib import oauth2_app
from aiohttp_session import SimpleCookieStorage, get_session, setup as session_setup



async def index(request: web.Request):
    session = await get_session(request)
    return web.json_response({"user": session.get("user")})


async def logout(request: web.Request):
    session = await get_session(request)
    session.invalidate()
    return web.HTTPTemporaryRedirect(location="/")


async def on_facebook_login(request: web.Request, facebook_token):
    session = await get_session(request)

    async with request.app["session"].get(
            "https://graph.facebook.com/me?fields=email",
            headers={"Authorization": f"Bearer {facebook_token['access_token']}"},
    ) as r:
        session["user"] = await r.json()

    return web.HTTPTemporaryRedirect(location="/")


def app_factory() -> web.Application:
    app = web.Application()

    session_setup(app, SimpleCookieStorage())

    app.add_subapp(
        "/auth/facebook/",
        oauth2_app(
            authorize_url="https://www.facebook.com/dialog/oauth",
            token_url="https://graph.facebook.com/oauth/access_token",
            client_id="***",
            client_secret="***",
            scopes=["email"],
            on_login=on_facebook_login,
        )
    )

    app.add_routes([web.get("/", index), web.get("/auth/logout", logout)])

    return app


if __name__ == "__main__":
    web.run_app(app_factory(), host="127.0.0.1")

github.py

from aiohttp import web
from aiohttp_oauth2.client.contrib import github
from aiohttp_session import SimpleCookieStorage, get_session, setup as session_setup


async def index(request: web.Request):
    session = await get_session(request)
    return web.json_response({"user": session.get("user"), "emails": session.get("emails")})


async def logout(request: web.Request):
    session = await get_session(request)
    session.invalidate()
    return web.HTTPTemporaryRedirect(location="/")


async def on_github_login(request: web.Request, github_token):
    session = await get_session(request)

    async with request.app["session"].get(
            "https://api.github.com/user",
            headers={"Authorization": f"Bearer {github_token['access_token']}"},
    ) as r:
        session["user"] = await r.json()

    async with request.app["session"].get(
            "https://api.github.com/user/emails",
            headers={"Authorization": f"Bearer {github_token['access_token']}"},
    ) as r:
        session["emails"] = await r.json()

    return web.HTTPTemporaryRedirect(location="/")


def app_factory() -> web.Application:
    app = web.Application()

    session_setup(app, SimpleCookieStorage())

    app.add_subapp(
        "/auth/github/",
        github(
            "****",
            "***",
            on_login=on_github_login,
            scopes=["user:email"],
        ),
    )

    app.add_routes([web.get("/", index), web.get("/auth/logout", logout)])

    return app


if __name__ == "__main__":
    web.run_app(app_factory(), host="127.0.0.1")