jenkins-bot submitted this change.
[IMPR] add a tiny cache wrapper
- the wrapper adds an attribute to the instance which holds the result
of the decorated method. The attribute's name is the method name with
preleading underscore.
- tests added
- cache some methods
Change-Id: I560873ce6c4092e75a3733599252a7992e5fc890
---
M pywikibot/comms/eventstreams.py
M pywikibot/flow.py
M pywikibot/logentries.py
M pywikibot/page/_pages.py
M pywikibot/page/_wikibase.py
M pywikibot/proofreadpage.py
M pywikibot/site/_basesite.py
M pywikibot/tools/__init__.py
M tests/tools_tests.py
9 files changed, 265 insertions(+), 145 deletions(-)
diff --git a/pywikibot/comms/eventstreams.py b/pywikibot/comms/eventstreams.py
index 1c82bc9..dd4d844 100644
--- a/pywikibot/comms/eventstreams.py
+++ b/pywikibot/comms/eventstreams.py
@@ -24,6 +24,7 @@
from requests.packages.urllib3.util.response import httplib
from pywikibot import Site, Timestamp, config, debug, warning
+from pywikibot.tools import cached
try:
@@ -137,23 +138,20 @@
'{}={!r}'.format(k, v) for k, v in kwargs.items()))
@property
+ @cached
def url(self):
"""Get the EventStream's url.
:raises NotImplementedError: no stream types specified
"""
- if not hasattr(self, '_url'):
- if self._streams is None:
- raise NotImplementedError(
- 'No streams specified for class {}'
- .format(self.__class__.__name__))
- self._url = ('{host}{path}/{streams}{since}'
- .format(host=self._site.eventstreams_host(),
- path=self._site.eventstreams_path(),
- streams=self._streams,
- since=('?since={}'.format(self._since)
- if self._since else '')))
- return self._url
+ if self._streams is None:
+ raise NotImplementedError('No streams specified for class {}'
+ .format(self.__class__.__name__))
+ return '{host}{path}/{streams}{since}'.format(
+ host=self._site.eventstreams_host(),
+ path=self._site.eventstreams_path(),
+ streams=self._streams,
+ since='?since={}'.format(self._since) if self._since else '')
def set_maximum_items(self, value: int) -> None:
"""
diff --git a/pywikibot/flow.py b/pywikibot/flow.py
index 88f7082..3cc747a 100644
--- a/pywikibot/flow.py
+++ b/pywikibot/flow.py
@@ -18,6 +18,7 @@
UnknownExtensionError,
)
from pywikibot.page import BasePage, PageSourceType, User
+from pywikibot.tools import cached
logger = logging.getLogger('pywiki.wiki.flow')
@@ -55,14 +56,13 @@
raise NotImplementedError
@property
+ @cached
def uuid(self) -> str:
"""Return the UUID of the page.
:return: UUID of the page
"""
- if not hasattr(self, '_uuid'):
- self._uuid = self._load()['workflowId']
- return self._uuid
+ return self._load()['workflowId']
def get(self, force: bool = False, get_redirect: bool = False
) -> Dict[str, Any]:
diff --git a/pywikibot/logentries.py b/pywikibot/logentries.py
index 75463a7..33d36df 100644
--- a/pywikibot/logentries.py
+++ b/pywikibot/logentries.py
@@ -12,6 +12,7 @@
import pywikibot
from pywikibot.backports import Dict, List, Tuple
from pywikibot.exceptions import Error, HiddenKeyError
+from pywikibot.tools import cached
class LogEntry(UserDict):
@@ -100,22 +101,19 @@
return self[self._expected_type] # old behaviour
return self.get('params', {})
+ @cached
def page(self) -> Union[int, 'pywikibot.page.Page']:
"""
Page on which action was performed.
:return: page on action was performed
"""
- if not hasattr(self, '_page'):
- self._page = pywikibot.Page(self.site, self['title'])
- return self._page
+ return pywikibot.Page(self.site, self['title'])
+ @cached
def timestamp(self) -> 'pywikibot.Timestamp':
"""Timestamp object corresponding to event timestamp."""
- if not hasattr(self, '_timestamp'):
- self._timestamp = pywikibot.Timestamp.fromISOformat(
- self['timestamp'])
- return self._timestamp
+ return pywikibot.Timestamp.fromISOformat(self['timestamp'])
class OtherLogEntry(LogEntry):
@@ -127,6 +125,7 @@
"""A log entry whose target is a user page."""
+ @cached
def page(self) -> 'pywikibot.page.User':
"""Return the target user.
@@ -135,9 +134,7 @@
:return: target user
"""
- if not hasattr(self, '_page'):
- self._page = pywikibot.User(self.site, self['title'])
- return self._page
+ return pywikibot.User(self.site, self['title'])
class BlockEntry(LogEntry):
@@ -176,6 +173,7 @@
return super().page()
+ @cached
def flags(self) -> List[str]:
"""
Return a list of (str) flags associated with the block entry.
@@ -186,36 +184,29 @@
"""
if self.action() == 'unblock':
return []
- if not hasattr(self, '_flags'):
- self._flags = self._params.get('flags', [])
- # pre mw 1.25 returned a delimited string.
- if isinstance(self._flags, str):
- self._flags = self._flags.split(',') if self._flags else []
- return self._flags
+ flags = self._params.get('flags', [])
+ # pre mw 1.25 returned a delimited string.
+ if isinstance(flags, str):
+ flags = flags.split(',') if flags else []
+ return flags
+
+ @cached
def duration(self) -> Optional[datetime.timedelta]:
"""
Return a datetime.timedelta representing the block duration.
:return: datetime.timedelta, or None if block is indefinite.
"""
- if not hasattr(self, '_duration'):
- if self.expiry() is None:
- self._duration = None
- else:
- # Doing the difference is easier than parsing the string
- self._duration = self.expiry() - self.timestamp()
- return self._duration
+ # Doing the difference is easier than parsing the string
+ return (self.expiry() - self.timestamp()
+ if self.expiry() is not None else None)
+ @cached
def expiry(self) -> Optional['pywikibot.Timestamp']:
"""Return a Timestamp representing the block expiry date."""
- if not hasattr(self, '_expiry'):
- details = self._params.get('expiry')
- if details:
- self._expiry = pywikibot.Timestamp.fromISOformat(details)
- else:
- self._expiry = None # for infinite blocks
- return self._expiry
+ details = self._params.get('expiry')
+ return pywikibot.Timestamp.fromISOformat(details) if details else None
class RightsEntry(LogEntry):
@@ -249,11 +240,10 @@
_expected_type = 'upload'
+ @cached
def page(self) -> 'pywikibot.page.FilePage':
"""Return FilePage on which action was performed."""
- if not hasattr(self, '_page'):
- self._page = pywikibot.FilePage(self.site, self['title'])
- return self._page
+ return pywikibot.FilePage(self.site, self['title'])
class MoveEntry(LogEntry):
@@ -279,11 +269,10 @@
else self._params['new_title'])
@property
+ @cached
def target_page(self) -> 'pywikibot.page.Page':
"""Return target page object."""
- if not hasattr(self, '_target_page'):
- self._target_page = pywikibot.Page(self.site, self.target_title)
- return self._target_page
+ return pywikibot.Page(self.site, self.target_title)
def suppressedredirect(self) -> bool:
"""Return True if no redirect was created during the move."""
diff --git a/pywikibot/page/_pages.py b/pywikibot/page/_pages.py
index aee95a2..6d2f056 100644
--- a/pywikibot/page/_pages.py
+++ b/pywikibot/page/_pages.py
@@ -51,6 +51,7 @@
from pywikibot.page._links import BaseLink, Link
from pywikibot.site import Namespace, NamespaceArgType
from pywikibot.tools import (
+ cached,
ComparableMixin,
first_upper,
issue_deprecation_warning,
@@ -191,17 +192,14 @@
return self._contentmodel
@property
- def depth(self):
- """Return the depth/subpage level of the page."""
- if not hasattr(self, '_depth'):
- # Check if the namespace allows subpages
- if self.namespace().subpages:
- self._depth = self.title().count('/')
- else:
- # Does not allow subpages, which means depth is always 0
- self._depth = 0
+ @cached
+ def depth(self) -> int:
+ """Return the depth/subpage level of the page.
- return self._depth
+ Check if the namespace allows subpages.
+ Not allowed subpages means depth is always 0.
+ """
+ return self.title().count('/') if self.namespace().subpages else 0
@property
def pageid(self) -> int:
@@ -357,6 +355,7 @@
return self.site.base_url(
self.site.articlepath.format(self.title(as_url=True)))
+ @cached
def autoFormat(self):
"""
Return :py:obj:`date.getAutoFormat` dictName and value, if any.
@@ -367,12 +366,7 @@
different namespaces, as some sites have categories with the
same names. Regular titles return (None, None).
"""
- if not hasattr(self, '_autoFormat'):
- self._autoFormat = date.getAutoFormat(
- self.site.lang,
- self.title(with_ns=False)
- )
- return self._autoFormat
+ return date.getAutoFormat(self.site.lang, self.title(with_ns=False))
def isAutoTitle(self):
"""Return True if title of this Page is in the autoFormat dict."""
@@ -706,6 +700,7 @@
"""Return True if last editor was unregistered."""
return self.latest_revision.anon
+ @cached
def lastNonBotUser(self) -> str:
"""
Return name or IP address of last human/non-bot user to edit page.
@@ -717,16 +712,11 @@
i.e. which is not returned by Site.botusers(), it will be returned
as a non-bot edit.
"""
- if hasattr(self, '_lastNonBotUser'):
- return self._lastNonBotUser
-
- self._lastNonBotUser = None
for entry in self.revisions():
if entry.user and (not self.site.isBot(entry.user)):
- self._lastNonBotUser = entry.user
- break
+ return entry.user
- return self._lastNonBotUser
+ return None
def editTime(self) -> pywikibot.Timestamp:
"""Return timestamp of last revision to page."""
@@ -2146,6 +2136,7 @@
super().__init__(source, title, ns)
@property
+ @cached
def raw_extracted_templates(self):
"""
Extract templates using :py:obj:`textlib.extract_templates_and_params`.
@@ -2153,16 +2144,9 @@
Disabled parts and whitespace are stripped, except for
whitespace in anonymous positional arguments.
- This value is cached.
-
:rtype: list of (str, OrderedDict)
"""
- if not hasattr(self, '_raw_extracted_templates'):
- templates = textlib.extract_templates_and_params(
- self.text, True, True)
- self._raw_extracted_templates = templates
-
- return self._raw_extracted_templates
+ return textlib.extract_templates_and_params(self.text, True, True)
def templatesWithParams(self):
"""
diff --git a/pywikibot/page/_wikibase.py b/pywikibot/page/_wikibase.py
index e6fda16..0a342bc 100644
--- a/pywikibot/page/_wikibase.py
+++ b/pywikibot/page/_wikibase.py
@@ -46,6 +46,7 @@
from pywikibot.page._filepage import FilePage
from pywikibot.page._pages import BasePage
from pywikibot.site import DataSite, Namespace
+from pywikibot.tools import cached
__all__ = (
@@ -1167,11 +1168,10 @@
self._type = datatype
@property
+ @cached
def type(self) -> str:
"""Return the type of this property."""
- if not hasattr(self, '_type'):
- self._type = self.repo.getPropertyType(self)
- return self._type
+ return self.repo.getPropertyType(self)
def getID(self, numeric: bool = False):
"""
diff --git a/pywikibot/proofreadpage.py b/pywikibot/proofreadpage.py
index 9a6267c..9b0650b 100644
--- a/pywikibot/proofreadpage.py
+++ b/pywikibot/proofreadpage.py
@@ -50,6 +50,7 @@
from pywikibot.data.api import Request
from pywikibot.exceptions import Error, OtherPageSaveError
from pywikibot.page import PageSourceType
+from pywikibot.tools import cached
try:
@@ -557,6 +558,7 @@
return '/* {0.status} */ '.format(self)
@property
+ @cached
def url_image(self) -> str:
"""Get the file url of the scan of ProofreadPage.
@@ -567,35 +569,33 @@
:raises ValueError: in case of no prp_page_image src found for scan
"""
# wrong link fails with various possible Exceptions.
- if not hasattr(self, '_url_image'):
+ if self.exists():
+ url = self.full_url()
+ else:
+ path = 'w/index.php?title={}&action=edit&redlink=1'
+ url = self.site.base_url(path.format(self.title(as_url=True)))
- if self.exists():
- url = self.full_url()
- else:
- path = 'w/index.php?title={}&action=edit&redlink=1'
- url = self.site.base_url(path.format(self.title(as_url=True)))
+ try:
+ response = http.fetch(url, charset='utf-8')
+ except Exception:
+ pywikibot.error('Error fetching HTML for {}.'.format(self))
+ raise
- try:
- response = http.fetch(url, charset='utf-8')
- except Exception:
- pywikibot.error('Error fetching HTML for {}.'.format(self))
- raise
+ soup = _bs4_soup(response.text) # type: ignore
- soup = _bs4_soup(response.text) # type: ignore
+ try:
+ url_image = soup.find(class_='prp-page-image')
+ # if None raises AttributeError
+ url_image = url_image.find('img')
+ # if None raises TypeError.
+ url_image = url_image['src']
+ except (TypeError, AttributeError):
+ raise ValueError('No prp-page-image src found for {}.'
+ .format(self))
+ else:
+ url_image = 'https:' + url_image
- try:
- self._url_image = soup.find(class_='prp-page-image')
- # if None raises AttributeError
- self._url_image = self._url_image.find('img')
- # if None raises TypeError.
- self._url_image = self._url_image['src']
- except (TypeError, AttributeError):
- raise ValueError('No prp-page-image src found for {}.'
- .format(self))
- else:
- self._url_image = 'https:' + self._url_image
-
- return self._url_image
+ return url_image
def _ocr_callback(self, cmd_uri: str,
parser_func: Optional[Callable[[str], str]] = None,
diff --git a/pywikibot/site/_basesite.py b/pywikibot/site/_basesite.py
index 85da2fa..1bcc7f7 100644
--- a/pywikibot/site/_basesite.py
+++ b/pywikibot/site/_basesite.py
@@ -22,6 +22,7 @@
from pywikibot.site._namespace import Namespace, NamespacesDict
from pywikibot.throttle import Throttle
from pywikibot.tools import (
+ cached,
ComparableMixin,
SelfCallString,
first_upper,
@@ -98,11 +99,10 @@
self._locked_pages = set()
@property
+ @cached
def throttle(self):
"""Return this Site's throttle. Initialize a new one if needed."""
- if not hasattr(self, '_throttle'):
- self._throttle = Throttle(self)
- return self._throttle
+ return Throttle(self)
@property
def family(self):
@@ -128,36 +128,31 @@
return self.__code
@property
- def doc_subpage(self):
- """
- Return the documentation subpage for this Site.
+ @cached
+ def doc_subpage(self) -> tuple:
+ """Return the documentation subpage for this Site."""
+ try:
+ doc, codes = self.family.doc_subpages.get('_default', ((), []))
+ if self.code not in codes:
+ try:
+ doc = self.family.doc_subpages[self.code]
+ # Language not defined in doc_subpages in x_family.py file
+ # It will use default for the family.
+ # should it just raise an Exception and fail?
+ # this will help to check the dictionary ...
+ except KeyError:
+ warn('Site {} has no language defined in '
+ 'doc_subpages dict in {}_family.py file'
+ .format(self, self.family.name),
+ FamilyMaintenanceWarning, 2)
+ # doc_subpages not defined in x_family.py file
+ except AttributeError:
+ doc = () # default
+ warn('Site {} has no doc_subpages dict in {}_family.py file'
+ .format(self, self.family.name),
+ FamilyMaintenanceWarning, 2)
- :rtype: tuple
- """
- if not hasattr(self, '_doc_subpage'):
- try:
- doc, codes = self.family.doc_subpages.get('_default', ((), []))
- if self.code not in codes:
- try:
- doc = self.family.doc_subpages[self.code]
- # Language not defined in doc_subpages in x_family.py file
- # It will use default for the family.
- # should it just raise an Exception and fail?
- # this will help to check the dictionary ...
- except KeyError:
- warn('Site {} has no language defined in '
- 'doc_subpages dict in {}_family.py file'
- .format(self, self.family.name),
- FamilyMaintenanceWarning, 2)
- # doc_subpages not defined in x_family.py file
- except AttributeError:
- doc = () # default
- warn('Site {} has no doc_subpages dict in {}_family.py file'
- .format(self, self.family.name),
- FamilyMaintenanceWarning, 2)
- self._doc_subpage = doc
-
- return self._doc_subpage
+ return doc
def _cmpkey(self):
"""Perform equality and inequality tests on Site objects."""
@@ -245,11 +240,10 @@
return Namespace.builtin_namespaces()
@property
+ @cached
def namespaces(self):
"""Return dict of valid namespaces on this wiki."""
- if not hasattr(self, '_namespaces'):
- self._namespaces = NamespacesDict(self._build_namespaces())
- return self._namespaces
+ return NamespacesDict(self._build_namespaces())
def ns_normalize(self, value):
"""
diff --git a/pywikibot/tools/__init__.py b/pywikibot/tools/__init__.py
index e5f80c0..04fd27d 100644
--- a/pywikibot/tools/__init__.py
+++ b/pywikibot/tools/__init__.py
@@ -29,6 +29,7 @@
import pkg_resources
import pywikibot # T306760
+from pywikibot.backports import Callable
from pywikibot.tools._deprecate import ( # noqa: F401 skipcq: PY-W2000
ModuleDeprecationWrapper,
add_decorated_full_name,
@@ -1319,3 +1320,50 @@
bytes_to_read -= len(read_bytes)
sha.update(read_bytes)
return sha.hexdigest()
+
+
+def cached(*arg: Callable) -> Any:
+ """Decorator to cache information of an object.
+
+ The wrapper adds an attribute to the instance which holds the result
+ of the decorated method. The attribute's name is the method name
+ with preleading underscore.
+
+ Usage::
+
+ @cached
+ def this_method(self)
+
+ @cached
+ def that_method(self, force=False)
+
+ No parameter may be used with this decorator. Only a force parameter
+ may be used with the decorated method. All other parameters are
+ discarded and lead to a TypeError.
+
+ .. note:: A property must be decorated on top of the property method
+ below other decorators. This decorator must not be used with
+ functions.
+ .. versionadded:: 7.3
+
+ :raises TypeError: decorator must be used without arguments
+ """
+ fn = arg and arg[0]
+ if not callable(fn):
+ raise TypeError(
+ '"cached" decorator must be used without arguments.') from None
+
+ @wraps(fn)
+ def wrapper(obj: object, *, force=False) -> Any:
+ cache_name = '_' + fn.__name__
+ if force:
+ with suppress(AttributeError):
+ delattr(obj, cache_name)
+ try:
+ return getattr(obj, cache_name)
+ except AttributeError:
+ val = fn(obj)
+ setattr(obj, cache_name, val)
+ return val
+
+ return wrapper
diff --git a/tests/tools_tests.py b/tests/tools_tests.py
index db010d3..2dfb4a6 100755
--- a/tests/tools_tests.py
+++ b/tests/tools_tests.py
@@ -17,6 +17,7 @@
from pywikibot import tools
from pywikibot.tools import (
+ cached,
classproperty,
has_module,
intersect_generators,
@@ -924,6 +925,112 @@
tools.strtobool('okay')
+class DecoratedMethods:
+
+ """Test class to verify cached decorator."""
+
+ def __init__(self):
+ """Initializer, reset read counter."""
+ self.read = 0
+
+ @cached
+ def foo(self):
+ """A method."""
+ self.read += 1
+ return 'foo'
+
+ @property
+ @cached
+ def bar(self):
+ """A property."""
+ self.read += 1
+ return 'bar'
+
+ def baz(self):
+ """An undecorated method."""
+ self.read += 1
+ return 'baz'
+
+ @cached
+ def quux(self, force=False):
+ """Method with force."""
+ self.read += 1
+ return 'quux'
+
+ @cached
+ def method_with_args(self, *args, **kwargs):
+ """Method with force."""
+ self.read += 1
+ return 'method_with_args'
+
+
+class TestTinyCache(TestCase):
+
+ """Test cached decorator."""
+
+ net = False
+
+ def setUp(self):
+ """Setup tests."""
+ self.foo = DecoratedMethods()
+ super().setUp()
+
+ def test_cached(self):
+ """Test for cached decorator."""
+ self.assertEqual(self.foo.foo(), 'foo') # check computed value
+ self.assertEqual(self.foo.read, 1)
+ self.assertTrue(hasattr(self.foo, '_foo'))
+ self.assertEqual(self.foo.foo(), 'foo') # check cached value
+ self.assertEqual(self.foo.read, 1) # bar() was called only once
+ del self.foo._foo
+ self.assertFalse(hasattr(self.foo, '_foo'))
+ self.assertEqual(self.foo.foo(), 'foo') # check computed value
+ self.assertEqual(self.foo.__doc__,
+ 'Test class to verify cached decorator.')
+ self.assertEqual(self.foo.foo.__doc__, 'A method.')
+
+ def test_cached_property(self):
+ """Test for cached property decorator."""
+ self.assertEqual(self.foo.bar, 'bar')
+ self.assertEqual(self.foo.read, 1)
+ self.assertTrue(hasattr(self.foo, '_bar'))
+ self.assertEqual(self.foo.bar, 'bar')
+ self.assertEqual(self.foo.read, 1)
+
+ def test_cached_with_paramters(self):
+ """Test for cached decorator with parameters."""
+ msg = '"cached" decorator must be used without arguments'
+ with self.assertRaisesRegex(TypeError, msg):
+ cached(42)(self.foo.baz())
+ with self.assertRaisesRegex(TypeError, msg):
+ cached()(self.foo.baz())
+
+ def test_cached_with_force(self):
+ """Test for cached decorator with force enabled."""
+ self.assertEqual(self.foo.quux(), 'quux')
+ self.assertEqual(self.foo.read, 1)
+ self.assertTrue(hasattr(self.foo, '_quux'))
+ self.assertEqual(self.foo.quux(force=True), 'quux')
+ self.assertEqual(self.foo.read, 2)
+
+ def test_cached_with_argse(self):
+ """Test method with args."""
+ self.assertEqual(self.foo.method_with_args(force=False),
+ 'method_with_args')
+ self.assertEqual(self.foo.read, 1)
+ self.assertTrue(hasattr(self.foo, '_method_with_args'))
+ with self.assertRaises(TypeError):
+ self.foo.method_with_args(True)
+ with self.assertRaises(TypeError):
+ self.foo.method_with_args(bar='baz')
+ with self.assertRaises(TypeError):
+ self.foo.method_with_args(1, 2, foo='bar')
+ self.assertEqual(self.foo.method_with_args(force=True),
+ 'method_with_args')
+ self.assertEqual(self.foo.method_with_args(), 'method_with_args')
+ self.assertEqual(self.foo.read, 2)
+
+
if __name__ == '__main__': # pragma: no cover
with suppress(SystemExit):
unittest.main()
To view, visit change 423951. To unsubscribe, or for help writing mail filters, visit settings.