-
Notifications
You must be signed in to change notification settings - Fork 0
/
api.py
83 lines (71 loc) · 2.97 KB
/
api.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
import requests
import urllib3
from kom import KOM
def get_strava_auth_url(client_id: str, redirect_uri: str) -> str:
return (
f"https://www.strava.com/oauth/authorize?client_id={client_id}"
f"&redirect_uri={redirect_uri}"
f"&response_type=code&scope=activity:read_all"
)
def get_refresh_token_from_auth_code(client_id: str, client_secret: str, auth_code: str) -> str:
url = "https://www.strava.com/oauth/token"
payload = {
"client_id": client_id,
"client_secret": client_secret,
"code": auth_code,
}
response = requests.post(url, data=payload).json()
return response["refresh_token"]
def get_access_token_from_refresh_token(client_id: str, client_secret: str, refresh_token: str) -> str:
url = "https://www.strava.com/oauth/token"
payload = {
"client_id": client_id,
"client_secret": client_secret,
"refresh_token": refresh_token,
"grant_type": "refresh_token",
"f": "json",
}
response = requests.post(url, data=payload, verify=False).json()
return response["access_token"]
def get_access_token_from_auth_code(client_id: str, client_secret: str, auth_code: str) -> str:
refresh_token = get_refresh_token_from_auth_code(client_id, client_secret, auth_code)
return get_access_token_from_refresh_token(client_id, client_secret, refresh_token)
def make_strava_api_request(url, access_token, params={}):
# Prevents "Unverified HTTPS request is being made to host 'www.strava.com'"
# warning
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
header = {"Authorization": "Bearer " + access_token}
response = requests.get(url, headers=header, params=params)
if response.status_code != 200:
raise Exception(
f"Strava API request failed with status code {response.status_code}. "
f"Response: {response.text}"
)
return response.json()
def get_all_pages(url: str, access_token: str, per_page: int = 200):
page_num = 1
all_responses = []
while True:
responses = make_strava_api_request(
url, access_token, params={"per_page": per_page, "page": page_num}
)
if len(responses):
all_responses.extend(responses)
page_num += 1
else:
# All pages have been pulled
break
return all_responses
def get_athlete_koms(access_token: str) -> "list[KOM]":
# TODO try building swagger_client library
# import swagger_client
# swagger_client.configuration.access_token = access_token
# api_instance = swagger_client.ActivitiesApi()
athlete_url = "https://www.strava.com/api/v3/athlete"
athlete_info = make_strava_api_request(athlete_url, access_token)
athlete_id = athlete_info["id"]
kom_responses = get_all_pages(
url=f"https://www.strava.com/api/v3/athletes/{athlete_id}/koms",
access_token=access_token,
)
return [KOM(post_response) for post_response in kom_responses]