mediawiki_dump_tools/Mediawiki-Utilities/mw/util/api.py

   1 import logging
   2 import time
   3
   4 import requests
   5
   6 from .functions import none_or
   7
   8 logger = logging.getLogger("mw.util.api.session")
   9
  10 FAILURE_THRESHOLD = 5
  11 TIMEOUT = 20
  12
  13
  14 class Session:
  15     def __init__(self, uri, headers=None, timeout=None,
  16                  failure_threshold=None, wait_step=2):
  17         if uri is None:
  18             raise TypeError("uri must not be None")
  19
  20         self.uri = str(uri)
  21         self.headers = headers if headers is not None else {}
  22         self.session = requests.Session()
  23
  24         self.failure_threshold = int(failure_threshold or FAILURE_THRESHOLD)
  25         self.timeout = float(TIMEOUT)
  26         self.wait_step = float(wait_step)
  27
  28         self.failed = 0
  29
  30     def __sleep(self):
  31         time.sleep(self.failed * (self.wait_step ** self.failed))
  32
  33     def get(self, params, **kwargs):
  34         return self.request('GET', params, **kwargs)
  35
  36     def post(self, params, **kwargs):
  37         return self.request('POST', params, **kwargs)
  38
  39     def request(self, type, params):
  40         try:
  41             result = self.session.request(type, self.uri, params=params,
  42                                           timeout=self.timeout)
  43             self.failed = 0
  44             return result
  45         except (requests.HTTPError, requests.ConnectionError):
  46             self.failed += 1
  47
  48             if self.failed > self.failure_threshold:
  49                 self.failed = 0
  50                 raise
  51             else:
  52                 self.__sleep()
  53                 self.request(type, params)