jenkins-bot has submitted this change. (
https://gerrit.wikimedia.org/r/c/pywikibot/core/+/836251 )
Change subject: [IMPR] Add APISite.ratelimit() method
......................................................................
[IMPR] Add APISite.ratelimit() method
- add a RateLimit namedtuple to tools/collections.py
- add APISite.ratelimit() method
- use it in api.Request instead of private _ratelimited() method
to handle ratelimited warning
- add some tests
Bug: T304808
Change-Id: I5db248dc4ef4bea03ad3ac4589830c209b961a04
---
M pywikibot/site/_apisite.py
M pywikibot/data/api/_requests.py
M pywikibot/tools/collections.py
M tests/site_tests.py
4 files changed, 204 insertions(+), 16 deletions(-)
Approvals:
JJMC89: Looks good to me, approved
jenkins-bot: Verified
diff --git a/pywikibot/data/api/_requests.py b/pywikibot/data/api/_requests.py
index 3b63cb2..916fc41 100644
--- a/pywikibot/data/api/_requests.py
+++ b/pywikibot/data/api/_requests.py
@@ -918,21 +918,11 @@
return True
def _ratelimited(self) -> None:
- """Handle ratelimited warning."""
- ratelimits = self.site.userinfo['ratelimits']
- delay = None
+ """Handle ratelimited warning.
- ratelimit = ratelimits.get(self.action, {})
- # find the lowest wait time for the given action
- for limit in ratelimit.values():
- seconds = limit['seconds']
- hits = limit['hits']
- delay = min(delay or seconds, seconds / hits)
-
- if not delay:
- pywikibot.warning(
- f'No rate limit found for action {self.action}')
- self.wait(delay)
+ This is also called from :meth:`_default_warning_handler`.
+ """
+ self.wait(self.site.ratelimit(self.action).delay)
def _bad_token(self, code) -> bool:
"""Check for bad token.
diff --git a/pywikibot/site/_apisite.py b/pywikibot/site/_apisite.py
index 1b4b1e6..2135184 100644
--- a/pywikibot/site/_apisite.py
+++ b/pywikibot/site/_apisite.py
@@ -78,6 +78,7 @@
merge_unique_dicts,
normalize_username,
)
+from pywikibot.tools.collections import RateLimit
if TYPE_CHECKING:
@@ -85,8 +86,8 @@
__all__ = ('APISite', )
-_mw_msg_cache: DefaultDict[str, dict[str, str]] = defaultdict(dict)
+_mw_msg_cache: DefaultDict[str, dict[str, str]] = defaultdict(dict)
_CompType = Union[int, str, 'pywikibot.page.Page', 'pywikibot.page.Revision']
_RequestWrapperT = TypeVar('_RequestWrapperT', bound='api._RequestWrapper')
@@ -498,6 +499,108 @@
return int(parameter['limit']) # T78333, T161783
+ def ratelimit(self, action: str) -> RateLimit:
+ """Get the rate limit for a given action.
+
+ This method get the ratelimit for a given action and returns a
+ :class:`tools.collections.RateLimit` namedtuple which has the
+ following fields and properties:
+
+ - ``group`` --- The current user group returned by the API. If
+ the user is not logged in, the group will be 'ip'.
+ - ``hits`` --- rate limit hits; API requests should not exceed
+ this limit value for the given action.
+ - ``seconds`` --- time base in seconds for the maximum hits
+ - ``delay`` --- *(property)* calculated as seconds per hits
+ which may be used for wait cycles.
+ - ``ratio`` --- *(property)* inverse of delay, calculated as
+ hits per seconds. The result may be Infinite.
+
+ If the user has 'noratelimit' rights, :meth:`maxlimit` is used
+ for ``hits`` and ``seconds`` will be 0. 'noratelimit' is
+ returned as group parameter in that case.
+
+ If no rate limit is found for the given action, :meth:`maxlimit`
+ is used for ``hits`` and ``seconds`` will be
+ :ref:`config.put_throttle<Settings to Avoid Server Overload>`.
+ As group parameter 'unknown' is returned in that case.
+
+ **Examples:**
+
+ This is an example for a bot user which is not logged in. The
+ rate limit user group is 'ip'
+
+ >>> site = pywikibot.Site()
+ >>> limit = site.ratelimit('edit') # get rate limit for 'edit' action
+ >>> limit
+ RateLimit(group='ip', hits=8, seconds=60)
+ >>> limit.delay # delay and ratio must be get as attributes
+ 7.5
+ >>> site.ratelimit('purge').hits # get purge hits
+ 30
+ >>> group, *limit = site.ratelimit('urlshortcode')
+ >>> group # the user is not logged in, we get 'ip' as group
+ 'ip'
+ >>> limit # starred assignment is allowed for the fields
+ [10, 120]
+
+ After login to the site and the rate limit will change. The
+ limit user group might be 'user':
+
+ >>> limit = site.ratelimit('edit') # doctest: +SKIP
+ >>> limit # doctest: +SKIP
+ RateLimit(group='user', hits=90, seconds=60)
+ >>> limit.ratio # doctest: +SKIP
+ 1.5
+ >>> limit = site.ratelimit('urlshortcode') # no action limit found
+ >>> group, *limits = limit
+ >>> group # doctest: +SKIP
+ 'unknown' # the group is 'unknown' because action was not found
+ >>> limits # doctest: +SKIP
+ (50, 10) # hits is maxlimit and seconds is config.put_throttle
+ >>> site.maxlimit, pywikibot.config.put_throttle
+ (50, 10)
+
+ If a user is logged in and has no rate limit, e.g bot accounts,
+ we always get a default RateLimit namedtuple like this:
+
+ >>> site.has_right['noratelimit'] # doctest: +SKIP
+ True
+ >>> limit = site.ratelimit('any_action') # maxlimit is used
+ >>> limit # doctest: +SKIP
+ RateLimit(group='noratelimit', hits=500, seconds=0)
+ >>> limit.delay, limit.ratio # doctest: +SKIP
+ (0.0, inf)
+
+ .. seealso:: :class:`tools.collections.RateLimit` for RateLimit
+ examples.
+ .. note:: It is not verified whether ``action`` parameter has a
+ valid value.
+ .. seealso:: :api:`Ratelimit`
+ .. versionadded:: 9.0
+
+ :param action: action which might be limited
+ :return: RateLimit tuple with ``group``, ``hits`` and ``seconds``
+ fields and properties for ``delay`` and ``ratio``.
+ """
+ ratio = 0
+ for key, value in self.userinfo['ratelimits'].get(action, {}).items():
+ h, s = value['hits'], value['seconds']
+ # find the highest ratio hits per seconds
+ if h / s > ratio:
+ limit = value
+ limit['group'] = key
+ ratio = h / s
+
+ if not ratio: # no limits found
+ limit = {'hits': self.maxlimit}
+ if self.has_right('noratelimit'):
+ limit['group'] = 'noratelimit'
+ else:
+ limit['seconds'] = pywikibot.config.put_throttle
+
+ return RateLimit(**limit)
+
@property
def userinfo(self) -> dict[str, Any]:
"""Retrieve userinfo from site and store in _userinfo attribute.
diff --git a/pywikibot/tools/collections.py b/pywikibot/tools/collections.py
index 63fbe43..38e330a 100644
--- a/pywikibot/tools/collections.py
+++ b/pywikibot/tools/collections.py
@@ -11,7 +11,7 @@
from collections.abc import Collection, Generator, Iterator, Mapping
from contextlib import suppress
from itertools import chain
-from typing import Any
+from typing import Any, NamedTuple
from pywikibot.backports import Generator as GeneratorType
@@ -21,6 +21,7 @@
'DequeGenerator',
'EmptyDefault',
'GeneratorWrapper',
+ 'RateLimit',
'SizedKeyCollection',
'EMPTY_DEFAULT',
)
@@ -294,3 +295,55 @@
"""Restart the generator."""
with suppress(AttributeError):
del self._started_gen
+
+
+class RateLimit(NamedTuple):
+
+ """A namedtuple which can hold rate limit content.
+
+ This class is used by :meth:`APISite.ratelimit()
+ <pywikibot.site._apisite.APISite.ratelimit>`.
+
+ .. note:: :meth:`delay` and :meth:`ratio` properties cannot be
+ sliced or used with tuple indices. The must be used as attributes.
+
+ >>> limit = RateLimit('user', 500, 10)
+ >>> limit
+ RateLimit(group='user', hits=500, seconds=10)
+ >>> limit.delay
+ 0.02
+ >>> limit.ratio
+ 50.0
+ >>> limit._fields
+ ('group', 'hits', 'seconds')
+ >>> limit._asdict() # doctest: +SKIP
+ {'group': 'user', 'hits': 500, 'seconds': 10}
+ >>> limit[0]
+ 'user'
+ >>> limit[-1]
+ 10
+ >>> user, hits, seconds = limit
+ >>> hits, seconds
+ (500, 10)
+ >>> newlimit = limit._replace(seconds=0)
+ >>> newlimit.delay
+ 0.0
+ >>> newlimit.ratio
+ inf
+
+ .. versionadded:: 9.0
+ """
+
+ group: str = 'unknown'
+ hits: int = 50
+ seconds: int = 0
+
+ @property
+ def delay(self) -> float:
+ """Calculate a delay value which is the inverse of :meth:`ratio`."""
+ return self.seconds / self.hits
+
+ @property
+ def ratio(self) -> float:
+ """Calculate a ratio how many hits can be done within one second."""
+ return self.hits / self.seconds if self.seconds != 0 else float('inf')
diff --git a/tests/site_tests.py b/tests/site_tests.py
index d2a51cf..840b5b2 100755
--- a/tests/site_tests.py
+++ b/tests/site_tests.py
@@ -277,6 +277,32 @@
if a:
self.assertEqual(a[0], mainpage)
+ def test_maxlimit(self):
+ """Test maxlimit property."""
+ limit = self.site.maxlimit
+ self.assertIsInstance(limit, int)
+ self.assertIn(limit, [10, 50, 500, 5000])
+
+ def test_ratelimit(self):
+ """Test ratelimit method."""
+ actions = ('edit', 'move', 'purge', 'invalid')
+ if self.site.logged_in():
+ groups = ['user', 'unknown', 'noratelimit']
+ else:
+ groups = ['ip', 'unknown']
+ self.assertFalse(self.site.has_right('noratelimit'))
+ for action in actions:
+ with self.subTest(action=action):
+ limit = self.site.ratelimit(action)
+ self.assertIn(limit.group, groups)
+ self.assertEqual(limit.seconds / limit.hits, limit.delay)
+ self.assertEqual(
+ 1 / limit.delay if limit.seconds else float('inf'),
+ limit.ratio)
+ if limit.group == 'unknown':
+ self.assertEqual(limit.hits, self.site.maxlimit)
+ self.assertEqual(limit.seconds, config.put_throttle)
+
class TestLockingPage(DefaultSiteTestCase):
"""Test cases for lock/unlock a page within threads."""
--
To view, visit https://gerrit.wikimedia.org/r/c/pywikibot/core/+/836251
To unsubscribe, or for help writing mail filters, visit
https://gerrit.wikimedia.org/r/settings
Gerrit-Project: pywikibot/core
Gerrit-Branch: master
Gerrit-Change-Id: I5db248dc4ef4bea03ad3ac4589830c209b961a04
Gerrit-Change-Number: 836251
Gerrit-PatchSet: 19
Gerrit-Owner: Xqt <[email protected]>
Gerrit-Reviewer: JJMC89 <[email protected]>
Gerrit-Reviewer: jenkins-bot
Gerrit-CC: Mpaa <[email protected]>
Gerrit-MessageType: merged
_______________________________________________
Pywikibot-commits mailing list -- [email protected]
To unsubscribe send an email to [email protected]
