from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import List, Tuple, Optional
from django.utils.timezone import now
from akvo.rsr.models import IatiActivityValidationJob, IatiOrganisationValidationJob
[docs]@dataclass(frozen=True)
class RequestRate:
count: int
period: timedelta
even_pace: bool = False
[docs]class RateLimiter:
''' Limit request to IATI validator API based on the subscription model
The rate limits can be configured according to the limits specified in
the subscription model used.
https://developer.iatistandard.org/subscriptions
'''
def __init__(self, rates: List[RequestRate]):
self.rates = rates
[docs] def is_allowed(self) -> bool:
for rate in self.rates:
if not self._apply_rate(rate):
return False
return True
def _apply_rate(self, rate: RequestRate) -> bool:
if rate.count == 0 or rate.period.total_seconds() == 0:
return True
started_at = now() - rate.period
count, latest = self._get_started_jobs_summary(started_at)
if count >= rate.count:
return False
if not rate.even_pace:
return True
# Spread the requests evenly in the given time period
spacing = timedelta(seconds=rate.period.total_seconds() / rate.count)
if latest and latest >= now() - spacing:
return False
return True
def _get_started_jobs_summary(self, started_at: datetime) -> Tuple[int, Optional[datetime]]:
activity_jobs = IatiActivityValidationJob.objects.filter(started_at__gt=started_at)
organisation_jobs = IatiOrganisationValidationJob.objects.filter(started_at__gt=started_at)
job_count = activity_jobs.count() + organisation_jobs.count()
datetimes = [j.started_at for j in [
activity_jobs.order_by('-started_at').first(),
organisation_jobs.order_by('-started_at').first()
] if j]
latest = max(datetimes) if datetimes else None
return (job_count, latest)