Spaces:
Sleeping
Sleeping
File size: 3,442 Bytes
af9f1a9 1f531ce af9f1a9 637ba4d be4141a af9f1a9 637ba4d bd3b857 af9f1a9 637ba4d af9f1a9 1f531ce af9f1a9 1f531ce a0871dc 1f531ce 99c4f8c a0871dc 99c4f8c 637ba4d af9f1a9 637ba4d af9f1a9 637ba4d af9f1a9 637ba4d af9f1a9 637ba4d af9f1a9 637ba4d af9f1a9 637ba4d af9f1a9 637ba4d 3975e9b af9f1a9 637ba4d af9f1a9 637ba4d af9f1a9 637ba4d af9f1a9 637ba4d af9f1a9 637ba4d af9f1a9 |
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 |
import logging
from fastapi import FastAPI
from strava_mcp.api import StravaAPI
from strava_mcp.config import StravaSettings
from strava_mcp.models import Activity, DetailedActivity, SegmentEffort
logger = logging.getLogger(__name__)
class StravaService:
"""Service for interacting with the Strava API."""
def __init__(self, settings: StravaSettings, app: FastAPI | None = None):
"""Initialize the Strava service.
Args:
settings: Strava API settings
app: FastAPI app for auth routes (optional)
"""
self.settings = settings
self.api = StravaAPI(settings, app)
async def initialize(self):
"""Initialize the service."""
# Log info about OAuth flow if no refresh token
if not self.settings.refresh_token:
logger.info(
"No STRAVA_REFRESH_TOKEN found in environment. "
"The standalone OAuth flow will be triggered automatically when needed."
)
async def close(self):
"""Close the API client."""
await self.api.close()
async def get_activities(
self,
before: int | None = None,
after: int | None = None,
page: int = 1,
per_page: int = 30,
) -> list[Activity]:
"""Get a list of activities for the authenticated athlete.
Args:
before: An epoch timestamp for filtering activities before a certain time
after: An epoch timestamp for filtering activities after a certain time
page: Page number
per_page: Number of items per page
Returns:
List of activities
"""
try:
logger.info("Getting activities for authenticated athlete")
activities = await self.api.get_activities(before, after, page, per_page)
logger.info(f"Retrieved {len(activities)} activities")
return activities
except Exception as e:
logger.error(f"Error getting activities: {str(e)}")
raise
async def get_activity(self, activity_id: int, include_all_efforts: bool = False) -> DetailedActivity:
"""Get a specific activity.
Args:
activity_id: The ID of the activity
include_all_efforts: Whether to include all segment efforts
Returns:
The activity details
"""
try:
logger.info(f"Getting activity {activity_id}")
activity = await self.api.get_activity(activity_id, include_all_efforts)
logger.info(f"Retrieved activity: {activity.name}")
return activity
except Exception as e:
logger.error(f"Error getting activity {activity_id}: {str(e)}")
raise
async def get_activity_segments(self, activity_id: int) -> list[SegmentEffort]:
"""Get segments from a specific activity.
Args:
activity_id: The ID of the activity
Returns:
List of segment efforts for the activity
"""
try:
logger.info(f"Getting segments for activity {activity_id}")
segments = await self.api.get_activity_segments(activity_id)
logger.info(f"Retrieved {len(segments)} segments")
return segments
except Exception as e:
logger.error(f"Error getting segments for activity {activity_id}: {str(e)}")
raise
|