timestream_1min_schedule.py

import time
import requests
import logging
import boto3
from botocore.config import Config


def send_price_data_to_timestream(write_client):
    base_url = "https://min-api.cryptocompare.com/data"
    symbols = "BTC,ETH,REP,DASH"
    r = requests.get(f"{base_url}/pricemulti?fsyms={symbols}&tsyms=USD")
    data = r.json()

    # extract the price for each cryptocurrency
    btc = data['BTC']['USD']
    eth = data['ETH']['USD']
    rep = data['REP']['USD']
    dash = data['DASH']['USD']

    # Convert data to the proper Timestream format: time, measure, dimension
    now = str(round(time.time() * 1000))
    current_prices = [{'Time': now,
                       'TimeUnit': 'MILLISECONDS',
                       'Dimensions': [{'Name': 'crypto', 'Value': 'BTC'}],
                       'MeasureName': 'Price',
                       'MeasureValue': str(btc),
                       'MeasureValueType': 'DOUBLE'},
                      {'Time': now,
                       'TimeUnit': 'MILLISECONDS',
                       'Dimensions': [{'Name': 'crypto', 'Value': 'ETH'}],
                       'MeasureName': 'Price',
                       'MeasureValue': str(eth),
                       'MeasureValueType': 'DOUBLE'},
                      {'Time': now,
                       'TimeUnit': 'MILLISECONDS',
                       'Dimensions': [{'Name': 'crypto', 'Value': 'DASH'}],
                       'MeasureName': 'Price',
                       'MeasureValue': str(dash),
                       'MeasureValueType': 'DOUBLE'},
                      {'Time': now,
                       'TimeUnit': 'MILLISECONDS',
                       'Dimensions': [{'Name': 'crypto', 'Value': 'REP'}],
                       'MeasureName': 'Price',
                       'MeasureValue': str(rep),
                       'MeasureValueType': 'DOUBLE'}]

    logger.info("Sending records to Timestream...")
    write_client.write_records(DatabaseName="demo",
                               TableName="demo",
                               Records=current_prices,
                               CommonAttributes={})
    logger.info("Price data written to Timestream.")


if __name__ == '__main__':
    logging.basicConfig(format="[%(levelname)s] [%(name)s] [%(asctime)s]: %(message)s", level="INFO")
    logger = logging.getLogger(__name__)
    session = boto3.Session(region_name='eu-central-1')
    timestream_write_client = session.client('timestream-write',
                                             config=Config(read_timeout=20,
                                                           max_pool_connections=5000,
                                                           retries={'max_attempts': 10}))

    for iteration in range(1, 7):
        logger.info("iteration nr %s", iteration)
        send_price_data_to_timestream(timestream_write_client)
        if iteration < 6:
            logger.info("Sleeping for 10 seconds...")
            time.sleep(10)