Source code for woob.tools.config.util

# Copyright(C) 2019-2021 Romain Bignon
#
# 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/>.


import os
from datetime import datetime

from woob.tools.log import getLogger


__all__ = ["LOGGER", "replace", "time_buffer"]


LOGGER = getLogger("woob.config")


try:
    from os import replace
except ImportError:

    def replace(src, dst, *args, **kwargs):
        if os.name != "posix":
            try:
                os.remove(dst)
            except OSError:
                pass
        return os.rename(src, dst, *args, **kwargs)


def time_buffer(since_seconds=None, last_run=True, logger=False):
    def decorator_time_buffer(func):
        def wrapper_time_buffer(*args, **kwargs):
            since_seconds = kwargs.pop("since_seconds", None)
            if since_seconds is None:
                since_seconds = decorator_time_buffer.since_seconds
            if logger:
                logger.debug(f"Time buffer for {func!r} of {since_seconds}. Last run {decorator_time_buffer.last_run}.")
            if since_seconds and decorator_time_buffer.last_run:
                if (datetime.now() - decorator_time_buffer.last_run).seconds < since_seconds:
                    if logger:
                        logger.debug("Too soon to run %r, ignore." % func)
                    return
            if logger:
                logger.debug("Run %r and record" % func)
            res = func(*args, **kwargs)
            decorator_time_buffer.last_run = datetime.now()
            return res

        decorator_time_buffer.since_seconds = since_seconds
        decorator_time_buffer.last_run = datetime.now() if last_run is True else last_run

        return wrapper_time_buffer

    return decorator_time_buffer