-
-
Notifications
You must be signed in to change notification settings - Fork 505
Expand file tree
/
Copy pathexample.py
More file actions
executable file
·174 lines (141 loc) · 5.73 KB
/
Copy pathexample.py
File metadata and controls
executable file
·174 lines (141 loc) · 5.73 KB
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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
#!/usr/bin/env python3
"""Simple Garmin Connect API Example.
Demonstrates authentication, token storage, and basic API calls.
For a comprehensive demo of all 127+ API methods, see demo.py
Dependencies:
pip install garminconnect[example]
pip install curl_cffi
Environment Variables (optional):
export GARMIN_EMAIL=<your garmin email address>
export GARMIN_PASSWORD=<your garmin password>
export GARMINTOKENS=<path to token storage, default ~/.garminconnect>
"""
import contextlib
import logging
import os
import sys
from datetime import date
from getpass import getpass
from pathlib import Path
from garminconnect import (
Garmin,
GarminConnectAuthenticationError,
GarminConnectConnectionError,
GarminConnectNotFoundError,
GarminConnectTooManyRequestsError,
)
logging.getLogger("garminconnect").setLevel(logging.CRITICAL)
def _status_code_from_error(exc: Exception) -> int | None:
"""Best-effort HTTP status extraction from a library exception.
The library attaches the underlying response object to many exceptions,
so use that first. Otherwise fall back to scanning the message string.
"""
response = getattr(exc, "response", None)
if response is not None:
status = getattr(response, "status_code", None)
if isinstance(status, int):
return status
error_str = str(exc)
for code in ("400", "401", "403", "404", "429", "500"):
if code in error_str:
return int(code)
return None
def safe_api_call(api_method, *args, **kwargs):
"""Call an API method and return (success, result, error_message)."""
try:
result = api_method(*args, **kwargs)
return True, result, None
except GarminConnectNotFoundError as e:
return False, None, f"Not found (404) — endpoint may have moved: {e}"
except GarminConnectAuthenticationError as e:
return False, None, f"Authentication error: {e}"
except GarminConnectTooManyRequestsError as e:
return False, None, f"Rate limit exceeded: {e}"
except GarminConnectConnectionError as e:
status = _status_code_from_error(e)
if status == 400:
return (
False,
None,
"Not available (400) — feature may not be enabled for your account",
)
if status == 401:
return False, None, "Authentication required (401) — please re-authenticate"
if status == 403:
return False, None, "Access denied (403) — account may not have permission"
if status == 404:
return False, None, "Not found (404) — endpoint may have moved"
if status == 429:
return False, None, "Rate limit (429) — please wait before retrying"
if status == 500:
return False, None, "Server error (500) — Garmin servers are having issues"
return False, None, f"Connection error: {e}"
except Exception as e:
return False, None, f"Unexpected error: {e}"
def init_api() -> Garmin | None:
"""Initialise Garmin API, restoring saved tokens or logging in fresh.
Tokens are stored in ``~/.garminconnect/garmin_tokens.json``
and reused automatically on the next run. DI OAuth tokens include
a refresh token so the session auto-renews without user interaction.
"""
tokenstore = os.getenv("GARMINTOKENS", "~/.garminconnect")
tokenstore_path = str(Path(tokenstore).expanduser())
# Try to restore saved tokens
try:
garmin = Garmin()
garmin.login(tokenstore_path)
print("Logged in using saved tokens.")
return garmin
except GarminConnectTooManyRequestsError as err:
print(f"Rate limit: {err}")
sys.exit(1)
except (GarminConnectAuthenticationError, GarminConnectConnectionError):
print("No valid tokens found — please log in.")
# Fresh credential login with MFA support
while True:
try:
email = os.getenv("GARMIN_EMAIL") or input("Email: ").strip()
password = os.getenv("GARMIN_PASSWORD") or getpass("Password: ")
garmin = Garmin(
email=email,
password=password,
prompt_mfa=lambda: input("MFA code: ").strip(),
)
# Don't keep the plaintext password in local scope longer than needed.
password = None
garmin.login(tokenstore_path)
print(f"Login successful. Tokens saved to: {tokenstore_path}")
return garmin
except GarminConnectTooManyRequestsError as err:
print(f"Rate limit: {err}")
sys.exit(1)
except GarminConnectAuthenticationError:
print("Wrong credentials — please try again.")
continue
except GarminConnectConnectionError as err:
print(f"Connection error: {err}")
return None
except KeyboardInterrupt:
return None
def main():
"""Basic usage example."""
api = init_api()
if not api:
return
today = date.today().isoformat()
success, summary, err = safe_api_call(api.get_user_summary, today)
if success and summary:
print(f"Steps today : {summary.get('totalSteps', 0)}")
print(f"Calories : {summary.get('totalKilocalories', 0):.0f} kcal")
dist_km = summary.get("totalDistanceMeters", 0) / 1000
print(f"Distance : {dist_km:.2f} km")
elif err:
print(f"Could not fetch summary: {err}")
success, hr, err = safe_api_call(api.get_heart_rates, today)
if success and hr:
print(f"Resting HR : {hr.get('restingHeartRate', 'n/a')} bpm")
elif err:
print(f"Could not fetch heart rate: {err}")
if __name__ == "__main__":
with contextlib.suppress(KeyboardInterrupt):
main()