import httpx import os import time from typing import Optional, Dict, Any, List from dotenv import load_dotenv load_dotenv() class StravaClient: def __init__(self): self.client_id = os.getenv("STRAVA_CLIENT_ID") self.client_secret = os.getenv("STRAVA_CLIENT_SECRET") self.refresh_token = os.getenv("STRAVA_REFRESH_TOKEN") self.access_token: Optional[str] = None self.expires_at: int = 0 self.base_url = "https://www.strava.com/api/v3" async def _refresh_access_token(self): """Refreshes the access token using the refresh token.""" async with httpx.AsyncClient() as client: response = await client.post( "https://www.strava.com/oauth/token", data={ "client_id": self.client_id, "client_secret": self.client_secret, "grant_type": "refresh_token", "refresh_token": self.refresh_token, }, ) response.raise_for_status() data = response.json() self.access_token = data["access_token"] self.refresh_token = data["refresh_token"] self.expires_at = data["expires_at"] async def get_valid_token(self) -> str: """Returns a valid access token, refreshing it if necessary.""" if not self.refresh_token: raise ValueError( "No Strava refresh token found. Please run 'uv run auth' on your local machine to authenticate first." ) if not self.access_token or time.time() > self.expires_at - 60: await self._refresh_access_token() return self.access_token async def _get(self, endpoint: str, params: Optional[Dict[str, Any]] = None) -> Any: """Helper for authenticated GET requests to the Strava API.""" token = await self.get_valid_token() async with httpx.AsyncClient() as client: response = await client.get( f"{self.base_url}/{endpoint}", headers={"Authorization": f"Bearer {token}"}, params=params, ) response.raise_for_status() return response.json() # ── Athletes ────────────────────────────────────────────────────────────── async def get_athlete(self) -> Dict[str, Any]: """Gets the current athlete profile.""" return await self._get("athlete") async def get_athlete_zones(self) -> Dict[str, Any]: """Gets heart rate and power zones configured for the current athlete.""" return await self._get("athlete/zones") async def get_athlete_stats(self) -> Dict[str, Any]: """Gets cumulative stats for the current athlete (totals for runs, rides, swims).""" athlete = await self.get_athlete() return await self._get(f"athletes/{athlete['id']}/stats") # ── Activities ──────────────────────────────────────────────────────────── async def list_activities( self, limit: int = 10, page: int = 1, before: int | None = None, after: int | None = None, ) -> List[Dict[str, Any]]: """Lists activities for the current athlete.""" params: Dict[str, Any] = {"per_page": limit, "page": page} if before is not None: params["before"] = before if after is not None: params["after"] = after return await self._get("athlete/activities", params=params) async def get_activity(self, activity_id: int) -> Dict[str, Any]: """Gets a specific activity by ID.""" return await self._get(f"activities/{activity_id}") async def get_activity_laps(self, activity_id: int) -> List[Dict[str, Any]]: """Gets the laps for a specific activity.""" return await self._get(f"activities/{activity_id}/laps") async def get_activity_zones(self, activity_id: int) -> List[Dict[str, Any]]: """Gets the heart rate and power zones for a specific activity.""" return await self._get(f"activities/{activity_id}/zones") async def get_activity_comments( self, activity_id: int, page: int = 1, per_page: int = 30 ) -> List[Dict[str, Any]]: """Gets comments on a specific activity.""" return await self._get( f"activities/{activity_id}/comments", params={"page": page, "per_page": per_page}, ) async def get_activity_kudoers( self, activity_id: int, page: int = 1, per_page: int = 30 ) -> List[Dict[str, Any]]: """Gets athletes who kudoed a specific activity.""" return await self._get( f"activities/{activity_id}/kudos", params={"page": page, "per_page": per_page}, ) async def get_activity_streams( self, activity_id: int, keys: List[str] ) -> Dict[str, Any]: """Gets data streams for a specific activity.""" return await self._get( f"activities/{activity_id}/streams", params={"keys": ",".join(keys), "key_by_type": True}, ) # ── Clubs ───────────────────────────────────────────────────────────────── async def get_athlete_clubs( self, page: int = 1, per_page: int = 30 ) -> List[Dict[str, Any]]: """Gets clubs the current athlete belongs to.""" return await self._get( "athlete/clubs", params={"page": page, "per_page": per_page} ) async def get_club(self, club_id: int) -> Dict[str, Any]: """Gets a specific club by ID.""" return await self._get(f"clubs/{club_id}") async def get_club_activities( self, club_id: int, page: int = 1, per_page: int = 30 ) -> List[Dict[str, Any]]: """Gets recent activities from a club.""" return await self._get( f"clubs/{club_id}/activities", params={"page": page, "per_page": per_page} ) async def get_club_members( self, club_id: int, page: int = 1, per_page: int = 30 ) -> List[Dict[str, Any]]: """Gets members of a club.""" return await self._get( f"clubs/{club_id}/members", params={"page": page, "per_page": per_page} ) # ── Routes ──────────────────────────────────────────────────────────────── async def get_routes_by_athlete( self, page: int = 1, per_page: int = 30 ) -> List[Dict[str, Any]]: """Gets routes created by the current athlete.""" athlete = await self.get_athlete() return await self._get( f"athletes/{athlete['id']}/routes", params={"page": page, "per_page": per_page}, ) async def get_route_by_id(self, route_id: str) -> Dict[str, Any]: """Gets a specific route by its ID.""" return await self._get(f"routes/{route_id}") async def get_route_streams(self, route_id: str) -> Dict[str, Any]: """Gets data streams for a specific route.""" return await self._get(f"routes/{route_id}/streams") # ── Segments ────────────────────────────────────────────────────────────── async def get_segment(self, segment_id: int) -> Dict[str, Any]: """Gets a specific segment by ID.""" return await self._get(f"segments/{segment_id}") async def get_starred_segments( self, page: int = 1, per_page: int = 30 ) -> List[Dict[str, Any]]: """Gets segments starred by the current athlete.""" return await self._get( "segments/starred", params={"page": page, "per_page": per_page} ) async def explore_segments( self, bounds: str, activity_type: str = "riding" ) -> Dict[str, Any]: """Explores segments within a given bounding box.""" return await self._get( "segments/explore", params={"bounds": bounds, "activity_type": activity_type}, ) async def get_segment_streams( self, segment_id: int, keys: List[str] ) -> Dict[str, Any]: """Gets data streams for a specific segment.""" return await self._get( f"segments/{segment_id}/streams", params={"keys": ",".join(keys), "key_by_type": True}, ) # ── Segment Efforts ─────────────────────────────────────────────────────── async def get_segment_effort(self, effort_id: str) -> Dict[str, Any]: """Gets a specific segment effort by ID.""" return await self._get(f"segment_efforts/{effort_id}") async def list_segment_efforts( self, segment_id: int, start_date_local: Optional[str] = None, end_date_local: Optional[str] = None, per_page: int = 30, ) -> List[Dict[str, Any]]: """Lists efforts on a segment for the current athlete.""" params: Dict[str, Any] = {"segment_id": segment_id, "per_page": per_page} if start_date_local: params["start_date_local"] = start_date_local if end_date_local: params["end_date_local"] = end_date_local return await self._get("segment_efforts", params=params) async def get_segment_effort_streams( self, effort_id: str, keys: List[str] ) -> Dict[str, Any]: """Gets data streams for a specific segment effort.""" return await self._get( f"segment_efforts/{effort_id}/streams", params={"keys": ",".join(keys), "key_by_type": True}, ) # ── Gear ────────────────────────────────────────────────────────────────── async def get_gear_by_id(self, gear_id: str) -> Dict[str, Any]: """Gets details for a specific piece of gear (bike or shoes) by its ID.""" return await self._get(f"gear/{gear_id}")