FMttMr6c

1from dataclasses import dataclass
2from typing import Any, Iterable
3import requests
4from pylimit import PyRateLimit
5import time
6
7KEY = 'YOUR KEY HERE'
8
9class Api:
10  def get_data(self, page, credentials):
11    params = {
12      'access_key': credentials,
13      'search': 'new york',
14      'limit': '50',
15      'offset': str((page - 1) * 50)
16    }
17    return requests.get('http://api.marketstack.com/v1/tickers/', params).json()
18
19api = Api()
20
21def limitor(namespace: str, fnn):
22    lim = PyRateLimit(period=1, limit=9)
23    sleep = [0.1, 0.4, 0.8, 1.3]
24    attempts = 0
25 
26    while not lim.attempt(namespace):
27        time.sleep(sleep.pop() if len(sleep) else 2)
28        if attempts > 20:
29            raise Exception(f"api limit exception {namespace}")
30 
31    return fnn()
32
33def handler(task:Task) -> TaskResult:
34    this_data = limitor(task.name + task.credentials, lambda: api.get_data(page = task.page, credentials = task.credentials))
35 
36    if this_data['pagination']['limit'] + this_data['pagination']['offset'] < this_data['pagination']['total']:
37        return TaskResult(task, this_data, [Task(dict(**task.__dict__ | {"page": task.page + 1}))])
38 
39    return TaskResult(task, this_data, [])
40
41@dataclass
42class Task:
43    name:str
44    credentials:str
45    page:int = 1
46
47@dataclass
48class TaskResult:
49    task:Task
50    data:Any
51    next_tasks: Iterable[Task]
52
53t = Task('Some task', KEY)
54handler(t)