Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
for ride in q:
try:
client = data.StravaClientForAthlete(ride.athlete)
# TODO: Make it configurable to force refresh of data.
activity_json = self.get_cached_activity_json(ride) if use_cache else None
if activity_json is None:
if options.only_cache:
self.logger.info("[CACHE-MISS] Skipping ride {} since there is no cached version.")
continue
self.logger.info("[CACHE-MISS] Fetching activity detail for {!r}".format(ride))
# We do this manually, so that we can cache the JSON for later use.
activity_json = client.protocol.get('/activities/{id}', id=ride.id, include_all_efforts=True)
strava_activity = stravamodel.Activity.deserialize(activity_json, bind_client=client)
try:
self.logger.info("Caching activity {!r}".format(ride))
self.cache_activity(strava_activity, activity_json)
except:
log.error("Error caching activity {} (ignoring)".format(strava_activity),
exc_info=self.logger.isEnabledFor(logging.DEBUG))
else:
strava_activity = stravamodel.Activity.deserialize(activity_json, bind_client=client)
self.logger.info("[CACHE-HIT] Using cached activity detail for {!r}".format(ride))
try:
self.logger.info("Writing out GPS track for {!r}".format(ride))
data.write_ride_track(strava_activity, ride)
except:
continue
self.logger.info("[CACHE-MISS] Fetching activity detail for {!r}".format(ride))
# We do this manually, so that we can cache the JSON for later use.
activity_json = client.protocol.get('/activities/{id}', id=ride.id, include_all_efforts=True)
strava_activity = stravamodel.Activity.deserialize(activity_json, bind_client=client)
try:
self.logger.info("Caching activity {!r}".format(ride))
self.cache_activity(strava_activity, activity_json)
except:
log.error("Error caching activity {} (ignoring)".format(strava_activity),
exc_info=self.logger.isEnabledFor(logging.DEBUG))
else:
strava_activity = stravamodel.Activity.deserialize(activity_json, bind_client=client)
self.logger.info("[CACHE-HIT] Using cached activity detail for {!r}".format(ride))
# try:
# self.logger.info("Writing out GPS track for {!r}".format(ride))
# data.write_ride_track(strava_activity, ride)
# except:
# self.logger.error("Error writing track for activity {0}, athlete {1}".format(ride.id, ride.athlete),
# exc_info=self.logger.isEnabledFor(logging.DEBUG))
# raise
# We do this just to take advantage of the use-cache/only-cache feature for reprocessing activities.
data.update_ride_from_activity(strava_activity=strava_activity, ride=ride)
meta.session_factory().flush()
try:
self.logger.info("Writing out efforts for {!r}".format(ride))
:return: An iterator of :class:`stravalib.model.Activity` objects.
:rtype: :class:`BatchedResultsIterator`
"""
if before:
before = self._utc_datetime_to_epoch(before)
if after:
after = self._utc_datetime_to_epoch(after)
params = dict(before=before, after=after)
result_fetcher = functools.partial(self.protocol.get,
'/athlete/activities',
**params)
return BatchedResultsIterator(entity=model.Activity,
bind_client=self,
result_fetcher=result_fetcher,
limit=limit)
continue
self.logger.info("[CACHE-MISS] Fetching activity detail for {!r}".format(ride))
# We do this manually, so that we can cache the JSON for later use.
activity_json = client.protocol.get('/activities/{id}', id=ride.id, include_all_efforts=True)
strava_activity = stravamodel.Activity.deserialize(activity_json, bind_client=client)
try:
self.logger.info("Caching activity {!r}".format(ride))
self.cache_activity(strava_activity, activity_json)
except:
log.error("Error caching activity {} (ignoring)".format(strava_activity),
exc_info=self.logger.isEnabledFor(logging.DEBUG))
else:
strava_activity = stravamodel.Activity.deserialize(activity_json, bind_client=client)
self.logger.info("[CACHE-HIT] Using cached activity detail for {!r}".format(ride))
try:
self.logger.info("Writing out GPS track for {!r}".format(ride))
data.write_ride_track(strava_activity, ride)
except:
self.logger.error("Error writing track for activity {0}, athlete {1}".format(ride.id, ride.athlete),
exc_info=self.logger.isEnabledFor(logging.DEBUG))
raise
try:
self.logger.info("Writing out efforts for {!r}".format(ride))
data.write_ride_efforts(strava_activity, ride)
except:
self.logger.error("Error writing efforts for activity {0}, athlete {1}".format(ride.id, ride.athlete),
exc_info=self.logger.isEnabledFor(logging.DEBUG))
activity_json = self.get_cached_activity_json(ride) if use_cache else None
if options.progress:
i += 1
bar.update(i)
if activity_json is None:
if options.only_cache:
self.logger.info("[CACHE-MISS] Skipping ride {} since there is no cached version.")
continue
self.logger.info("[CACHE-MISS] Fetching activity detail for {!r}".format(ride))
# We do this manually, so that we can cache the JSON for later use.
activity_json = client.protocol.get('/activities/{id}', id=ride.id, include_all_efforts=True)
strava_activity = stravamodel.Activity.deserialize(activity_json, bind_client=client)
try:
self.logger.info("Caching activity {!r}".format(ride))
self.cache_activity(strava_activity, activity_json)
except:
log.error("Error caching activity {} (ignoring)".format(strava_activity),
exc_info=self.logger.isEnabledFor(logging.DEBUG))
else:
strava_activity = stravamodel.Activity.deserialize(activity_json, bind_client=client)
self.logger.info("[CACHE-HIT] Using cached activity detail for {!r}".format(ride))
# try:
# self.logger.info("Writing out GPS track for {!r}".format(ride))
# data.write_ride_track(strava_activity, ride)
# except:
if trainer is not None:
params['trainer'] = int(trainer)
if gear_id is not None:
params['gear_id'] = gear_id
if description is not None:
params['description'] = description
if device_name is not None:
params['device_name'] = device_name
raw_activity = self.protocol.put('/activities/{activity_id}', activity_id=activity_id, **params)
return model.Activity.deserialize(raw_activity, bind_client=self)
)
)
return True
else:
return False
try:
activities = client.get_activities(
after=start_date, limit=None
) # type: List[stravalib.orm.Activity]
filtered_rides = [
a
for a in activities
if (
(
a.type == strava_model.Activity.RIDE
or a.type == strava_model.Activity.EBIKERIDE
)
and not a.manual
and not a.trainer
and not is_excluded(a)
)
]
except HTTPError as e:
if (
"access_token" in e.message
): # A bit of a kludge, but don't have a way of hooking into the response processing earlier.
raise InvalidAuthorizationToken(
"Invalid authrization token for {}".format(athlete)
)
# Otherwise just fall-through and re-raise same exception.
)
return True
else:
return False
try:
activities = client.get_activities(
after=start_date, limit=None
) # type: List[stravalib.orm.Activity]
filtered_rides = [
a
for a in activities
if (
(
a.type == strava_model.Activity.RIDE
or a.type == strava_model.Activity.EBIKERIDE
)
and not a.manual
and not a.trainer
and not is_excluded(a)
)
]
except HTTPError as e:
if (
"access_token" in e.message
): # A bit of a kludge, but don't have a way of hooking into the response processing earlier.
raise InvalidAuthorizationToken(
"Invalid authrization token for {}".format(athlete)
)
# Otherwise just fall-through and re-raise same exception.
raise e