Glue ETL Script Example for Loading Tulip Table Data

import sys
import pandas as pd
import numpy as np
import requests
import json
import boto3
from botocore.exceptions import ClientError
from sqlalchemy import create_engine
import sqlalchemy as sa
from sqlalchemy.engine.url import URL
import psycopg2
from datetime import datetime
import logging
logger = logging.getLogger()

table_id = 'aKzvoscgHCyd2CRu3_DEFAULT'
def get_secret(secret_name, region_name):
    # Create a Secrets Manager client
    session = boto3.session.Session()
    client = session.client(
        service_name='secretsmanager',
        region_name=region_name
    )
    try:
        get_secret_value_response = client.get_secret_value(
            SecretId=secret_name
        )
    except ClientError as e:
        # For a list of exceptions thrown, see
        # https://docs.aws.amazon.com/secretsmanager/latest/apireference/API_GetSecretValue.html
        raise e
    return json.loads(get_secret_value_response['SecretString'])

redshift_credentials = get_secret(secret_name='tulip_redshift', region_name='us-east-1')
api_credentials = get_secret(secret_name='[INSTANCE].tulip.co-API-KEY', region_name='us-east-1')

# build SQL engine
url = URL.create(
drivername='postgresql',
host=redshift_credentials['host'],
port=redshift_credentials['port'],
database=redshift_credentials['dbname'],
username=redshift_credentials['username'],
password=redshift_credentials['password']
)
engine = sa.create_engine(url)

header = {'Authorization' : api_credentials['auth_header']}
base_url = 'https://william.tulip.co/api/v3'


offset = 0
function = f'/tables/{table_id}/records?limit=100&offset={offset}&includeTotalCount=false&filterAggregator=all'
r = requests.get(base_url+function, headers=header)
df = pd.DataFrame(r.json())
length = len(r.json())

while length > 0:
    offset += 100
    function = f'/tables/{table_id}/records?limit=100&offset={offset}&includeTotalCount=false&filterAggregator=all'
    r = requests.get(base_url+function,
    headers=header)
    length = len(r.json())
    df_append = pd.DataFrame(r.json())
    df = pd.concat([df, df_append], axis=0)

# capture date-time stamp
now = datetime.now()
df['datetime_updated'] = now

# write to Redshift
df.to_sql('station_activity_from_glue', engine, schema='product_growth', index=False,if_exists='replace')