def decorator()

in images/airflow/2.9.2/python/mwaa/logging/utils.py [0:0]

• 28 lines of code
• 6 McCabe index (conditional complexity)

    def decorator(func: F) -> F:
        last_called_global = [0.0]

        @wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            if instance_level_throttling:
                self = args[0]
                if not hasattr(self, "_last_called"):
                    setattr(self, "_last_called", {})
                last_called = self._last_called.get(func.__name__, 0.0)
            else:
                last_called = last_called_global[0]

            current_time = time.time()
            elapsed = current_time - last_called
            if elapsed < seconds:
                wait_time = seconds - elapsed
                if log_throttling_msg:
                    print(
                        f"Throttling {func.__name__} for {wait_time:.2f} more seconds."
                    )
                return None
            else:
                if instance_level_throttling:
                    self = args[0]
                    self._last_called[func.__name__] = current_time
                else:
                    last_called_global[0] = current_time
                return func(*args, **kwargs)

        return wrapper  # type: ignore