"""Utilities
==============
Rate limiter classes.
These are basically callables that when called register that a request was
issued. Depending on how they are configured that may cause a pause or
exception if a rate limit has been exceeded. It is up to the calling
code to ensure that these callables are invoked with every (successful?) call
to the backend API.
TODO: There is probably a better way to hook these into the requests library
directly
From the Strava docs:
Strava API usage is limited on a per-application basis using a short term,
15 minute, limit and a long term, daily, limit. The default rate limit allows
600 requests every 15 minutes, with up to 30,000 requests per day.
This limit allows applications to make 40 requests per minute for about
half the day.
"""
from __future__ import annotations
import logging
import time
from collections.abc import Callable
from logging import Logger
from typing import Literal, NamedTuple
import arrow
from stravalib.protocol import RequestMethod
[docs]
class RequestRate(NamedTuple):
"""Tuple containing request usage and usage limit."""
short_usage: int
"""15-minute usage"""
long_usage: int
"""Daily usage"""
short_limit: int
"""15-minutes limit"""
long_limit: int
"""Daily limit"""
[docs]
def get_seconds_until_next_quarter(
now: arrow.arrow.Arrow | None = None,
) -> int:
"""Returns the number of seconds until the next quarter of an hour. This is
the short-term rate limit used by Strava.
Parameters
----------
now : arrow.arrow.Arrow
A (utc) timestamp
Returns
-------
int
The number of seconds until the next quarter, as int
"""
if now is None:
now = arrow.utcnow()
return (
899
- (
now
- now.replace(
minute=(now.minute // 15) * 15, second=0, microsecond=0
)
).seconds
)
[docs]
def get_seconds_until_next_day(now: arrow.arrow.Arrow | None = None) -> int:
"""Returns the number of seconds until the next day (utc midnight). This is
the long-term rate limit used by Strava.
Parameters
----------
now : arrow.arrow.Arrow
A (utc) timestamp
Returns
-------
Int
The number of seconds until next day, as int
"""
if now is None:
now = arrow.utcnow()
return (now.ceil("day") - now).seconds
[docs]
class SleepingRateLimitRule:
"""A rate limit rule that can be prioritized and can dynamically adapt its
limits based on API responses. Given its priority, it will enforce a
variable "cool-down" period after each response. When rate limits are
reached within their period, this limiter will wait until the end of that
period. It will NOT raise any kind of exception in this case.
"""
[docs]
def __init__(
self,
priority: Literal["low", "medium", "high"] = "high",
) -> None:
"""
Constructs a new SleepingRateLimitRule.
Parameters
----------
priority : Literal["low", "medium", "high"]
The priority for this rule. When 'low', the cool-down period
after each request will be such that the long-term limits will
not be exceeded. When 'medium', the cool-down period will be such
that the short-term limits will not be exceeded. When 'high',
there will be no cool-down period.
"""
if priority not in ["low", "medium", "high"]:
raise ValueError(
f'Invalid priority "{priority}", expecting one of "low", "medium" or "high"'
)
self.log = logging.getLogger(
f"{self.__class__.__module__}.{self.__class__.__name__}"
)
self.priority = priority
def _get_wait_time(
self,
rates: RequestRate,
seconds_until_short_limit: int,
seconds_until_long_limit: int,
) -> float:
"""Calculate how much time user has until they can make another
request"""
# If limits are exceeded, wait until they reset
if rates.long_usage >= rates.long_limit:
self.log.warning("Long term API rate limit exceeded")
return seconds_until_long_limit
elif rates.short_usage >= rates.short_limit:
self.log.warning("Short term API rate limit exceeded")
return seconds_until_short_limit
# High priority: no wait time
if self.priority == "high":
return 0
# Calculate wait times for BOTH limits
short_wait = seconds_until_short_limit / (
rates.short_limit - rates.short_usage
)
long_wait = seconds_until_long_limit / (
rates.long_limit - rates.long_usage
)
if self.priority == "medium":
# Focus on short-term limit, but also respect daily limit
# when at least half of the daily quota is used
if rates.long_usage >= rates.long_limit / 2:
return max(short_wait, long_wait)
else:
return short_wait
elif self.priority == "low":
# Spread requests over the day, but always respect both limits
return max(short_wait, long_wait)
def __call__(
self, response_headers: dict[str, str], method: RequestMethod
) -> None:
"""Determines wait time until a call can be made again"""
rates = get_rates_from_response_headers(response_headers, method)
self.log.debug(f"Throttling based on rates: {rates}")
if rates:
time.sleep(
self._get_wait_time(
rates,
get_seconds_until_next_quarter(),
get_seconds_until_next_day(),
)
)
else:
self.log.warning("No rates present in response headers")
[docs]
class RateLimiter:
[docs]
def __init__(self) -> None:
self.log: Logger = logging.getLogger(
f"{self.__class__.__module__}.{self.__class__.__name__}"
)
self.rules: list[Callable[[dict[str, str], RequestMethod], None]] = []
def __call__(self, args: dict[str, str], method: RequestMethod) -> None:
"""Register another request is being issued."""
for r in self.rules:
r(args, method)
[docs]
class DefaultRateLimiter(RateLimiter):
"""Implements something similar to the default rate limit for Strava apps.
See https://developers.strava.com/docs/rate-limits/ and
https://communityhub.strava.com/t5/developer-knowledge-base/our-developer-program/ta-p/8849.
Rate limits are enforced by throttling requests based on their method and
client/app-specific limits imposed by Strava.
The rate limiter supports three priority levels:
- **high**: No cool-down period between requests. Requests are made as fast
as possible until limits are reached, then waits until the limit period expires.
- **medium**: Applies a cool-down period to avoid exceeding short-term limits
(e.g., 600 requests per 15 minutes, actual limits are app-specific).
- **low**: Applies a cool-down period to avoid exceeding long-term limits
(e.g., 30,000 requests per day, actual limits are app-specific), spreading
requests evenly throughout the day.
Examples
--------
Using default (high priority) rate limiter::
from stravalib.client import Client
client = Client(access_token=token) # Uses high priority by default
Using a custom rate limiter with medium priority::
from stravalib.client import Client
from stravalib.util.limiter import DefaultRateLimiter
rate_limiter = DefaultRateLimiter(priority="medium")
client = Client(access_token=token, rate_limiter=rate_limiter)
"""
[docs]
def __init__(
self, priority: Literal["low", "medium", "high"] = "high"
) -> None:
"""
Initializes the rate limiter based on the given priority.
Parameters
----------
priority : Literal["low", "medium", "high"]
The priority given to the requests. Default is "high"
(i.e. no throttling).
"""
super().__init__()
self.rules.append(SleepingRateLimitRule(priority=priority))