Skip to content

Instantly share code, notes, and snippets.

@vrutik2809

vrutik2809/export_to_sheet.py

Created April 3, 2022 13:43
Show Gist options
  • Save vrutik2809/898b0d1f22ae0ef9933aa2d6cd82bbb3 to your computer and use it in GitHub Desktop.
Save vrutik2809/898b0d1f22ae0ef9933aa2d6cd82bbb3 to your computer and use it in GitHub Desktop.
Download ZIP
Exporting SQL table or schema to Google Sheets
Raw
export_to_sheet.py
import os.path
from sqlalchemy import create_engine, inspect, Table, MetaData
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
# If modifying these scopes, delete the file token.json.
SCOPES = ['https://www.googleapis.com/auth/spreadsheets']
# OAuth credential file name.
CLIENT_SECRETS_FILE = 'credentials.json'
# TODO replace variable with appropiate values
DB_URL = f"postgresql://{user}:{password}@{hostname}:{port}/{database-name}"
engine = create_engine(DB_URL)
def reflect_table(name, schema, engine, metadata=None, connection_to_use=None):
if metadata is None:
metadata = MetaData(bind=engine)
autoload_with = engine if connection_to_use is None else connection_to_use
return Table(name, metadata, schema=schema, autoload_with=autoload_with, extend_existing=True)
def generate_base_body_table(table_name,schema,engine,spreadsheet_name='test_spreadsheet'):
with engine.connect() as connection:
inspector = inspect(engine)
if table_name not in inspector.get_table_names(schema=schema):
raise Exception(f'Table "{schema + "." + table_name}" not found')
table = reflect_table(table_name, schema, engine, connection_to_use=connection)
spreadsheet_body = {}
spreadsheet_body['properties'] = {
'title': spreadsheet_name,
}
header_values = []
for col in table.columns:
header_values.append({
'userEnteredValue': {
'stringValue': col.name
}
})
spreadsheet_body['sheets'] = [
{
'properties': {
'title': table_name,
},
'data': {
'startRow': 0,
'startColumn': 0,
'rowData': [
{
'values': header_values
}
]
}
}
]
return spreadsheet_body
def generate_data_body_table(table_name,schema,engine):
with engine.connect() as connection:
inspector = inspect(engine)
if table_name not in inspector.get_table_names(schema=schema):
raise Exception(f'Table "{schema + "." + table_name}" not found')
table = reflect_table(table_name, schema, engine, connection_to_use=connection)
body = {}
body['values'] = []
for row in table.select().execute():
row_values = []
for col in table.columns:
row_values.append(str(row[col.name]))
body['values'].append(row_values)
return body
def export_table_to_google_sheet(client_secrets_file,scopes,table_name,schema,engine,spreadsheet_name='test_spreadsheet'):
creds = None
if os.path.exists('token.json'):
creds = Credentials.from_authorized_user_file('token.json', scopes)
if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token:
creds.refresh(Request())
else:
flow = InstalledAppFlow.from_client_secrets_file(
client_secrets_file, scopes)
creds = flow.run_local_server(port=8080)
with open('token.json', 'w') as token:
token.write(creds.to_json())
try:
service = build('sheets', 'v4', credentials=creds)
spreadsheet_body = generate_base_body_table(table_name, schema, engine, spreadsheet_name)
response = service.spreadsheets().create(body=spreadsheet_body).execute()
spreadsheet_id = response['spreadsheetId'] #
print(spreadsheet_id)
data_body = generate_data_body_table(table_name, schema, engine)
response = service.spreadsheets().values().append(spreadsheetId=spreadsheet_id, range=f'{table_name}!A2', valueInputOption='USER_ENTERED', insertDataOption='INSERT_ROWS', body=data_body).execute()
except HttpError as err:
print(err)
def generate_base_body_schema(schema,engine,spreadsheet_name='test_spreadsheet'):
with engine.connect() as connection:
inspector = inspect(engine)
if schema not in inspector.get_schema_names():
raise Exception(f'Schema "{schema}" not found')
tables = inspector.get_table_names(schema=schema)
spreadsheet_body = {}
spreadsheet_body['properties'] = {
'title': spreadsheet_name,
}
spreadsheet_body['sheets'] = []
for table in tables:
table_body = {}
table_body['properties'] = {
'title': table
}
header_values = []
for col in inspector.get_columns(table,schema=schema):
header_values.append({
'userEnteredValue': {
'stringValue': col['name']
}
})
table_body['data'] = {
'startRow': 0,
'startColumn': 0,
'rowData': [
{
'values': header_values
}
]
}
spreadsheet_body['sheets'].append(table_body)
return spreadsheet_body
def export_schema_to_google_sheet(client_secrets_file,scopes,schema,engine,spreadsheet_name='test_spreadsheet'):
creds = None
if os.path.exists('token.json'):
creds = Credentials.from_authorized_user_file('token.json', scopes)
if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token:
creds.refresh(Request())
else:
flow = InstalledAppFlow.from_client_secrets_file(
client_secrets_file, scopes)
creds = flow.run_local_server(port=8080)
with open('token.json', 'w') as token:
token.write(creds.to_json())
try:
service = build('sheets', 'v4', credentials=creds)
spreadsheet_body = generate_base_body_schema(schema, engine, spreadsheet_name)
response = service.spreadsheets().create(body=spreadsheet_body).execute()
spreadsheet_id = response['spreadsheetId'] #
print(spreadsheet_id)
with engine.connect() as connection:
inspector = inspect(engine)
for table_name in inspector.get_table_names(schema=schema):
data_body = generate_data_body_table(table_name, schema, engine)
response = service.spreadsheets().values().append(spreadsheetId=spreadsheet_id, range=f'{table_name}!A2', valueInputOption='USER_ENTERED', insertDataOption='INSERT_ROWS', body=data_body).execute()
except HttpError as err:
print(err)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment