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."""