Optimizing Strava data collection with Elastic APM and a custom script solution

from elasticsearch import Elasticsearch, helpers
from datetime import datetime, timedelta
import requests
import json


ELASTIC_PASSWORD = "password_for_strava-User"
CLOUD_ID = "Cloud_ID retrieved from cloud.elastic.co"
client = Elasticsearch(
   cloud_id=CLOUD_ID,
   basic_auth=("strava", ELASTIC_PASSWORD)
)
ES_INDEX = 'strava'


stravaBaseUrl = "https://www.strava.com/api/v3/"


payload = ""


headers = {
   "Authorization": "Bearer _authorization_token_acquired_from_strava"
}


def GetStravaActivities():
   url = stravaBaseUrl + "athlete/activities"
   querystring = {"per_page": "50", "page": "1"}
   #### ^^^above the `per_page` can be changed to a maxmimum of 50
   #### You can have 100 requests per 15 minutes, that would give you 1500 activities to retrieve in 15 minutes
   #### Maximum of 1.000 requests per day.
   #### after ever run don't forget to increase the page number.
   #### Don't forget that we are now calling a 2nd API for every activity.
   #### That means we can only collect 50 activites, since for every activite we call the streams API.
   return json.loads((requests.request(
       "GET", url, data=payload, headers=headers, params=querystring).text))


def GetStravaStreams(activity):
   #Detailed Streams API call
   url = stravaBaseUrl + "activities/" + str(activity['id']) + "/streams"
   querystring = {"keys":"time,distance,latlng,altitude,velocity_smooth,heartrate,cadence,watts,temp,moving,grade_smooth","key_by_type":"true"}
   streams = json.loads((requests.request("GET", url, data=payload, headers=headers, params=querystring)).text)
   # create the doc needed for the bulk request
   doc = {
       "_index": ES_INDEX,
       "_source": {
           "strava": activity,
           "data": {}
       }
   }
   for i in range(streams['time']['original_size']):
       # run the modification of the data in an extra function
       yield ModifyStravaStreams(doc, streams, activity, i)


def ModifyStravaStreams(doc, streams, activity, i):
   tempDateTime = datetime.strptime(activity['start_date'].replace('Z','')+activity['timezone'][4:10], '%Y-%m-%dT%H:%M:%S%z')
   for stream in streams:
       if stream == 'time':
           tempTime = tempDateTime + timedelta(seconds=streams['time']['data'][i])
           doc['_source']['@timestamp'] = tempTime.strftime('%Y-%m-%dT%H:%M:%S%z')
       elif stream == 'velocity_smooth':
           doc['_source']['data'][stream] = streams[stream]['data'][i]*3.6
       elif stream == 'latlng':
           doc['_source']['data'][stream] = {
               "lat": streams[stream]['data'][i][0],
               "lon": streams[stream]['data'][i][1]
           }
       else:
           doc['_source']['data'][stream] = streams[stream]['data'][i]
   return doc


def main():
   # get the activities
   activities = GetStravaActivities()
   for activity in activities:
       # I only care about cycling, you are free to modify this.
       if activity['type'] == "Ride" or activity['type'] == "VirtualRide":
           # I like to know what is going on and if an error occurs which activity it is.
           print(activity['upload_id'], activity['name'])
           helpers.bulk(client, GetStravaStreams(activity))
main()

Copy

import elasticapm

Copy

clientapm = elasticapm.Client(service_name="StravaToElasticsearch", server_url="https://url:443",
                             secret_token="...", environment='production')
elasticapm.instrument()

Copy

def main():
   # get the activities
   activities = GetStravaActivities()
   for activity in activities:
       # I only care about cycling, you are free to modify this.
       if activity['type'] == "Ride" or activity['type'] == "VirtualRide":
           # I like to know what is going on and if an error occurs which activity it is.
           print(activity['upload_id'], activity['name'])
           helpers.bulk(client, GetStravaStreams(activity))
main()

Copy

clientapm.begin_transaction(transaction_type="request")
clientapm.end_transaction(name="StravaActivities", result="success")

Copy

def main():
  # get the activities
  clientapm.begin_transaction(transaction_type="request")
  activities = GetStravaActivities()
  clientapm.end_transaction(name="StravaActivities", result="success")
  for activity in activities:
      # I only care about cycling, you are free to modify this.
      if activity['type'] == "Ride" or activity['type'] == "VirtualRide":
          # I like to know what is going on and if an error occurs which activity it is.
          print(activity['upload_id'], activity['name'])
          clientapm.begin_transaction(transaction_type="request")
          helpers.bulk(client, GetStravaStreams(activity))
          clientapm.end_transaction(name="StravaActivityData", result="success")

Copy