Get OAuth 2.0 Refresh Tokens

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}")

        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())

~example.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "055fc249-d766-45e6-97f0-6930587e5667",
   "metadata": {},
   "outputs": [],
   "source": [
    "from google.oauth2.credentials import Credentials\n",
    "\n",
    "credential = None\n",
    "try:\n",
    "    credential = Credentials.from_authorized_user_file(\"token.json\")\n",
    "except FileNotFoundError:\n",
    "    pass\n",
    "\n",
    "if credential is None:\n",
    "    SCOPE = 'https://www.googleapis.com/auth/gmail.readonly'\n",
    "    import json\n",
    "    with open(\"client_secret.json\") as f:\n",
    "        client = {k:v for k, v in json.load(f).items() if k in [\"auth_uri\", \"token_uri\", \"client_id\", \"client_secret\"]}\n",
    "    %run oauth2_authz_code_grant.py\n",
    "    grant_response = await oauth2_authz_code_grant(SCOPE, **client)\n",
    "    credential = Credentials.from_authorized_user_info(client | grant_response)\n",
    "    # save credential\n",
    "    s = credential.to_json([\"expiry\", \"access_token\"])\n",
    "    with open(\"token.json\", \"w\") as f:\n",
    "        f.write(s)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "604ada1a-ba75-409a-9919-384a07ec86dd",
   "metadata": {},
   "outputs": [],
   "source": [
    "from googleapiclient.discovery import build\n",
    "\n",
    "service = build('gmail', 'v1', credentials=credential)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fff33ed3-78d8-4e60-a752-8f1091719b66",
   "metadata": {},
   "outputs": [],
   "source": [
    "results = service.users().labels().list(userId='me').execute()\n",
    "labels = results.get('labels', [])\n",
    "\n",
    "if not labels:\n",
    "    print('No labels found.')\n",
    "else:\n",
    "    print('Labels:')\n",
    "    for label in labels:\n",
    "        print(label['name'])\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}