google ドライブの共有フォルダをAPIで作成する azechi/azechi#3

notebook.ipynb

oauth2_authz_code_grant.py

async def oauth2_authz_code_grant(scope, client_id, client_secret, auth_uri, token_uri, redirect_port=0, timeout=60):
    code, redirect_uri = await get_authorization_code(scope, client_id, auth_uri, redirect_port, timeout)
    return get_token(code, redirect_uri, client_id, client_secret, token_uri)


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}")
        from IPython.display import clear_output
        clear_output(True)
        try:
            await wait_for(event.wait(), timeout=timeout)
        except TimeoutError:
            print("timeout")
            code = None
    print("server closed")
    
    return code, redirect_uri

def get_token(code, redirect_uri, client_id, client_secret, token_uri):
    from urllib.parse import urlencode
    from urllib.request import urlopen, Request
    from json import loads

    params = urlencode({
        "client_id": client_id,
        "client_secret": client_secret,
        "grant_type": "authorization_code",
        "code": code,
        "redirect_uri": redirect_uri
    })

    req = Request(token_uri, params.encode('ascii'))
    with urlopen(req) as res:
        return loads(res.read())