Skip to content

Instantly share code, notes, and snippets.

@ofen
Created January 17, 2019 20:00
Show Gist options
  • Save ofen/3b182f61a052edf016d498463d77b008 to your computer and use it in GitHub Desktop.
Save ofen/3b182f61a052edf016d498463d77b008 to your computer and use it in GitHub Desktop.
Basic async Google Sheet api request based on aiohttp and asyncio
# Require PyJwt, AIOHTTP and PyCrypto (for RS256 in JWT)
import jwt
import time
import json
from pprint import pprint
import aiohttp
import asyncio
# Building JSW
def get_jwt():
with open('credentials.json') as f:
credentials = json.load(f)
iat = time.time()
# Token expiring in 1 hour
exp = iat + 3600
payload = {
'iss': credentials['client_email'],
'scope': 'https://www.googleapis.com/auth/spreadsheets',
'aud': 'https://www.googleapis.com/oauth2/v4/token',
'exp': exp,
'iat': iat
}
return jwt.encode(payload=payload, key=credentials['private_key'], algorithm='RS256').decode('utf8')
async def aio_read(spreadsheet_id, cell_range):
headers = {
'Content-Type': 'application/x-www-form-urlencoded'
}
payload = {
'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer',
'assertion': get_jwt(),
}
async with aiohttp.ClientSession() as session:
async with session.post('https://www.googleapis.com/oauth2/v4/token', headers=headers, data=payload) as res:
access_token = (await res.json())['access_token']
base_url = 'https://sheets.googleapis.com'
api_call = '/v4/spreadsheets/%s/values/%s' % (spreadsheet_id, cell_range)
headers = {
'Authorization': 'Bearer %s' % access_token,
'Content-Type': 'application/json'
}
async with session.get(base_url + api_call, headers=headers) as res:
return await res.json()
spreadsheet_id = 'your_spreadsheet_id'
# Task queue
tasks = asyncio.gather(
aio_read(spreadsheet_id, 'Тестовый список!A2'),
aio_read(spreadsheet_id, 'Тестовый список!A3'),
aio_read(spreadsheet_id, 'Тестовый список!A4'),
aio_read(spreadsheet_id, 'Тестовый список!A5'),
)
# Create eventloop and pass the tasks to it
loop = asyncio.get_event_loop()
result = loop.run_until_complete(tasks)
loop.close()
# Print result
pprint(result)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment