Jupyter NotebookでGMail APIにアクセスするためのOAuth 2.0 アクセストークンをAuthorization Code Flowで取得する

Jupyter Notebookで、一回きりのTCP接続の待ち受けをしてOAuth 2.0 のアクセストークンを取得する。

GMail APIに使用できるアクセストークンはデバイスフローでは取得できない。gmail関連のscopeを受け付けない。 Authorization Code Flow (認可コードフロー)のローカルリダイレクトを使ってNotebookで受け取る。

google api の場合、redirect_uriで localhostは事前に登録しなくても使用できる。ポート番号もauthorization リクエストのときにパラーメターで渡して決められるので、ランダムポートでよい。

Jupyter Notebookは既定でasyncioのループが実行されているので、asyncio.start_serverしてasyncio.wait_forで待てばいい。

async def get_authorization_code(scope, client_id, auth_uri, port=0, timeout=60):
    from asyncio import start_server, wait_for, Event, TimeoutError
    from urllib.parse import urlencode, urlparse, parse_qs

    event = Event()
    
    code: str
    async def handler(reader, writer):
        nonlocal code

        data = await reader.readline()
        data = data.decode()
        code = parse_qs(urlparse(data.split()[1]).query)["code"][0]
    
        writer.write(b"HTTP/1.1 200 OK\nContent-Length: 0\n\n")
        await writer.drain()
        writer.close()
        await writer.wait_closed()
        event.set()
    
    redirect_uri: str
    async with await start_server(handler, '0.0.0.0', port) as srv:
        _, port = srv.sockets[0].getsockname()
        redirect_uri = f"http://localhost:{port}"
        params = urlencode({
            "client_id": client_id,
            "scope": scope,
            "response_type": "code",
            "redirect_uri": redirect_uri
        })

        print(f"{auth_uri}?{params}")

        try:
            await wait_for(event.wait(), timeout=timeout)
        except TimeoutError:
            print("timeout")
            code = None

    print("server closed")
    return code, redirect_uri

認可コードを受信したときのレスポンスに204 No Contentではなくて200を返す。204ではブラウザで認証画面の表示が残ったままになってどうにも気持ち悪いから。