Source code for woob.browser.retry

# Copyright(C) 2017  Vincent A
#
# This file is part of woob.
#
# woob is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# woob is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with woob. If not, see <http://www.gnu.org/licenses/>.

from contextlib import contextmanager
from functools import wraps

from .browsers import LoginBrowser
from .exceptions import LoggedOut
from ..exceptions import BrowserUnavailable


__all__ = ['login_method', 'retry_on_logout', 'RetryLoginBrowser']


[docs]def login_method(func): """Decorate a method to indicate the browser is logging in. When the decorated method is called, pages like `woob.browser.pages.LoginPage` will not raise `LoggedOut`, since it is expected for the browser not to be logged yet. """ func.login_decorated = True @wraps(func) def wrapper(browser, *args, **kwargs): browser.logging_in += 1 try: return func(browser, *args, **kwargs) finally: browser.logging_in -= 1 return wrapper
[docs]def retry_on_logout(exc_check=LoggedOut, tries=4): """Decorate a function to retry several times in case of exception. The decorated function is called at max 4 times. It is retried only when it raises an exception of the type `woob.browser.exceptions.LoggedOut`. If the function call succeeds and returns an iterator, a wrapper to the iterator is returned. If iterating on the result raises a `LoggedOut`, the iterator is recreated by re-calling the function, but the values already yielded will not be re-yielded. For consistency, the function MUST always return values in the same order. Adding this decorator to a method which can be called from another decorated method should be avoided, since nested calls will greatly increase the number of retries. """ if not isinstance(exc_check, type) or not issubclass(exc_check, Exception): raise TypeError('retry_on_logout() must be called in order to decorate %r' % tries) def decorator(func): @wraps(func) def wrapper(browser, *args, **kwargs): cb = lambda: func(browser, *args, **kwargs) for i in range(tries, 0, -1): try: ret = cb() except exc_check as exc: browser.logger.info('%r raised, retrying', exc) continue if not (hasattr(ret, '__next__') or hasattr(ret, 'next')): return ret # simple value, no need to retry on items return iter_retry(cb, value=ret, remaining=i, exc_check=exc_check, logger=browser.logger) raise BrowserUnavailable('Site did not reply successfully after multiple tries') return wrapper return decorator
@contextmanager def retry_on_logout_context(tries=4, logger=None): for i in range(tries, 0, -1): try: yield except LoggedOut as exc: if logger: logger.debug('%r raised, retrying', exc) else: return raise BrowserUnavailable('Site did not reply successfully after multiple tries')
[docs]class RetryLoginBrowser(LoginBrowser): """Browser that can retry methods if the site logs out the session. Some sites can terminate a session anytime, redirecting to a login page. To avoid having to handle it in the middle of every method, one can simply let logouts raise a `woob.browser.exceptions.LoggedOut` exception that is handled with a retry, thanks to the `@retry_on_logout` decorator. The `woob.browser.pages.LoginPage` will raise `LoggedOut` if the browser is not currently logging in. To detect this situation, the `do_login` method MUST be decorated with `@login_method`. """ def __init__(self, *args, **kwargs): super(RetryLoginBrowser, self).__init__(*args, **kwargs) self.logging_in = 0 if not hasattr(self.do_login, 'login_decorated'): raise Exception('do_login method was not decorated with @login_method')
class iter_retry: # when the callback is retried, it will create a new iterator, but we may already yielded # some values, so we need to keep track of them and seek in the middle of the iterator def __init__(self, cb, remaining=4, value=None, exc_check=LoggedOut, logger=None): self.cb = cb self.it = iter(value) if value is not None else None self.items = [] self.remaining = remaining self.logger = logger self.exc_check = exc_check def __iter__(self): return self def __next__(self): if self.remaining <= 0: raise BrowserUnavailable('Site did not reply successfully after multiple tries') if self.it is None: self.it = iter(self.cb()) # recreated iterator, consume previous items try: nb = -1 for nb, sent in enumerate(self.items): new = next(self.it) if hasattr(new, 'iter_fields'): equal = dict(sent.iter_fields()) == dict(new.iter_fields()) else: equal = sent == new if not equal: # safety is not guaranteed raise BrowserUnavailable('Site replied inconsistently between retries, %r vs %r', sent, new) except StopIteration: raise BrowserUnavailable('Site replied fewer elements (%d) than last iteration (%d)', nb + 1, len(self.items)) except self.exc_check as exc: if self.logger: self.logger.info('%r raised, retrying', exc) self.it = None self.remaining -= 1 return next(self) # return one item try: obj = next(self.it) except self.exc_check as exc: if self.logger: self.logger.info('%r raised, retrying', exc) self.it = None self.remaining -= 1 return next(self) else: self.items.append(obj) return obj next = __next__