Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
streams_json = self.get_cached_streams_json(ride) if use_cache else None
if streams_json is None:
if options.only_cache:
self.logger.info("[CACHE-MISS] Skipping ride {} since there is no cached stream.".format(ride))
continue
self.logger.info("[CACHE-MISS] Fetching streams for {!r}".format(ride))
# We do this manually, so that we can cache the JSON for later use.
streams_json = client.protocol.get(
'/activities/{id}/streams/{types}'.format(id=ride.id, types='latlng,time,altitude'),
resolution='low'
)
streams = [stravamodel.Stream.deserialize(stream_struct, bind_client=client) for stream_struct in streams_json]
try:
self.logger.info("Caching streams for {!r}".format(ride))
self.cache_stream(ride, streams_json)
except:
log.error("Error caching streams for activity {} (ignoring)".format(ride),
exc_info=self.logger.isEnabledFor(logging.DEBUG))
else:
streams = [stravamodel.Stream.deserialize(stream_struct, bind_client=client) for stream_struct in streams_json]
self.logger.info("[CACHE-HIT] Using cached streams detail for {!r}".format(ride))
data.write_ride_streams(streams, ride)
meta.session_factory().commit()
except:
start_date_local = arrow.get(start_date_local).naive
params["start_date_local"] = start_date_local.strftime("%Y-%m-%dT%H:%M:%SZ")
if end_date_local:
if isinstance(end_date_local, six.string_types):
end_date_local = arrow.get(end_date_local).naive
params["end_date_local"] = end_date_local.strftime("%Y-%m-%dT%H:%M:%SZ")
if limit is not None:
params["limit"] = limit
result_fetcher = functools.partial(self.protocol.get,
'/segments/{segment_id}/all_efforts',
**params)
return BatchedResultsIterator(entity=model.BaseEffort, bind_client=self,
result_fetcher=result_fetcher, limit=limit)
def type(self):
if self._lazytype:
clazz = getattr(stravalib.model, self._lazytype)
else:
clazz = self._type
return clazz
def write_ride_efforts(strava_activity, ride):
"""
Writes out all effort associated with a ride to the database.
:param strava_activity: The :class:`stravalib.orm.Activity` that is associated with this effort.
:type strava_activity: :class:`stravalib.orm.Activity`
:param ride: The db model object for ride.
:type ride: :class:`bafs.orm.Ride`
"""
assert isinstance(strava_activity, strava_model.Activity)
assert isinstance(ride, Ride)
try:
# Start by removing any existing segments for the ride.
meta.engine.execute(
RideEffort.__table__.delete().where(
RideEffort.ride_id == strava_activity.id
)
)
# Then add them back in
for se in strava_activity.segment_efforts:
effort = RideEffort(
id=se.id,
ride_id=strava_activity.id,
elapsed_time=timedelta_to_seconds(se.elapsed_time),