Skip to content
Open
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
feat: enhance Garmin and Strava settings with improved token manageme…
…nt and environment variable support
  • Loading branch information
MarcChen committed Nov 15, 2025
commit f783f84599eaf31ad694df58558c1ca7ad163672
48 changes: 40 additions & 8 deletions garmin/utils.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,13 @@
import logging
import os
import sys
from datetime import datetime, timedelta
from pathlib import Path
from typing import List, Tuple

import garth
from garth.exc import GarthException, GarthHTTPError
from pydantic import Field
from pydantic import Field, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

logger = logging.getLogger(__name__)
Expand All @@ -15,11 +16,24 @@
class GarminSettings(BaseSettings):
"""Configuration settings for Garmin API client."""

garmin_username: str = Field(..., validation_alias="GARMIN_USERNAME")
garmin_password: str = Field(..., validation_alias="GARMIN_PASSWORD")
garmin_username: str = Field(
default=None,
alias="GARMIN_USERNAME",
description="Garmin Connect username only used to authenticate and retrieve tokens.", # noqa: E501
)
garmin_password: SecretStr = Field(
default=None,
alias="GARMIN_PASSWORD",
description="Garmin Connect password used for authentication.",
)
garmin_tokens_path: Path = Field(
default=Path(__file__).parent.parent / ".garth",
)
garmin_tokens: str = Field(
default=None,
alias="GARMIN_TOKENS",
description="Garmin Connect tokens from garth client.dumps() for authentication.", # noqa: E501
)

model_config = SettingsConfigDict(
env_file=Path(__file__).parent.parent / ".env", extra="ignore"
Expand All @@ -38,16 +52,25 @@ def get_credentials_for_garmin(garmin_settings: GarminSettings = GarminSettings(
"""
logger.info("Authenticating...")
try:
# First try to restore from environment variable
token_env = os.getenv("GARMIN_TOKENS")
if token_env:
garth.client.loads(token_env)
logger.info("Authenticated using GARMIN_TOKENS environment variable.")
return
# Fallback to username/password login
garth.login(
garmin_settings.garmin_username,
garmin_settings.garmin_password,
garmin_settings.garmin_password.get_secret_value(),
prompt_mfa=lambda: input("Enter MFA code: "),
)
garth.save(garmin_settings.garmin_tokens_path)
print()
logger.info("Successfully authenticated!")
except GarthHTTPError:
logger.info("Wrong credentials. Please check username and password.")
logger.info(
"Wrong credentials or authentication failed. Please check username and password." # noqa: E501
)
sys.exit(1)


Expand Down Expand Up @@ -103,7 +126,7 @@ def upload_fit_file_to_garmin(new_file_path: Path):
def list_virtual_cycling_activities(
last_n_days: int = 30,
) -> Tuple[List[str], List[datetime]]:
"""Return two lists: activity names and start times of virtual cycling activities from Garmin Connect."""
"""Return two lists: activity names and start times of virtual cycling activities from Garmin Connect.""" # noqa: E501
logger.info(
f"Retrieving virtual cycling activities from the last {last_n_days} days..."
)
Expand All @@ -127,13 +150,22 @@ def list_virtual_cycling_activities(
start_time = None
start_times.append(start_time)
logger.debug(
f"Found virtual cycling activity: {activity['activityName']} at {activity.get('startTimeLocal', '')} with elapsed time {activity.get('elapsedTime', '')}."
f"Found virtual cycling activity: {activity['activityName']} at {activity.get('startTimeLocal', '')} with elapsed time {activity.get('elapsedTime', '')}." # noqa: E501
)
return names, start_times


def dump_token_string_as_vars():
"""Utility function to dump Garmin tokens as environment variable strings."""
token_string = garth.client.dumps()
logger.info("Garmin token string (CAREFUL: save it securely!")
logger.info(token_string)


if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
authenticate_to_garmin()
# Example usage:
list_virtual_cycling_activities()
names, start_times = list_virtual_cycling_activities()
for name, start_time in zip(names, start_times):
print(f"Activity: {name}, Start Time: {start_time}")
54 changes: 41 additions & 13 deletions strava/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,11 +11,11 @@
import time
from datetime import datetime, timedelta
from pathlib import Path
from typing import List, Optional
from typing import Any, List, Optional
from urllib.parse import parse_qs, urlparse

import requests
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict
from requests import Session

Expand All @@ -25,19 +25,49 @@
class StravaSettings(BaseSettings):
"""Configuration settings for Strava API client."""

client_id: str = Field(..., validation_alias="CLIENT_ID")
client_secret: str = Field(..., validation_alias="CLIENT_SECRET")
client_id: str | None = Field(
default=None,
description="Strava API Client ID necessary to generate access tokens.",
)
client_secret: SecretStr | None = Field(
default=None,
description="Strava API Client Secret necessary to generate access tokens.",
)
token_type: str = Field(
default="Bearer",
)
access_token: SecretStr
expires_at: int
expires_in: int
refresh_token: SecretStr
token_url: str = "https://www.strava.com/oauth/token"
auth_base_url: str = "https://www.strava.com/oauth/authorize"
token_file: str = "strava_tokens.json"
cookie_file: str = "cookie.json"
token_file: Path = Path(__file__).parent.parent / "strava_tokens.json"
cookie_file: Path = Path(__file__).parent.parent / "cookie.json"
activities_url: str = "https://www.strava.com/api/v3/athlete/activities"
database_file: str = "strava.db"
database_file: Path = Path(__file__).parent.parent / "strava.db"

model_config = SettingsConfigDict(
env_file=Path(__file__).parent.parent / ".env", extra="ignore"
env_file=Path(__file__).parent.parent / ".env",
extra="ignore",
env_prefix="STRAVA_",
) # noqa: E501

def model_post_init(self, __context: Any) -> None:
"""Create necessary files if they don't exist."""
# Create token file if it doesn't exist
token_path = self.token_file
# Only dump selected fields to token file if it doesn't exist
if not token_path.exists():
token_data = {
"token_type": self.token_type,
"access_token": str(self.access_token.get_secret_value()),
"expires_at": self.expires_at,
"expires_in": self.expires_in,
"refresh_token": str(self.refresh_token.get_secret_value()),
}
token_path.write_text(json.dumps(token_data))


class TokenData(BaseModel):
"""Model for storing Strava API token data."""
Expand Down Expand Up @@ -170,7 +200,7 @@ def _fetch_token(self, redirect_url: str) -> None:
self.settings.token_url,
data={
"client_id": self.settings.client_id,
"client_secret": self.settings.client_secret,
"client_secret": str(self.settings.client_secret.get_secret_value()),
"code": code[0],
"grant_type": "authorization_code",
},
Expand Down Expand Up @@ -203,7 +233,7 @@ def refresh_token(self) -> None:
self.settings.token_url,
data={
"client_id": self.settings.client_id,
"client_secret": self.settings.client_secret,
"client_secret": str(self.settings.client_secret.get_secret_value()),
"grant_type": "refresh_token",
"refresh_token": self.token_data.refresh_token,
},
Expand Down Expand Up @@ -279,9 +309,7 @@ def _download_attempt(self, activity_id: int) -> bool:
activity_name = activity_data.get("name", f"activity_{activity_id}")
# Parse and trim start_date to only date part (YYYY-MM-DD)
start_date_str = activity_data.get("start_date", "unknown_date")
start_date = (
start_date_str.split("T")[0] if "T" in start_date_str else start_date_str
)
_ = start_date_str.split("T")[0] if "T" in start_date_str else start_date_str

# 2. Fetch time-series streams
stream_types = [
Expand Down