Skip to content

Проверка достижения любого лимита

limits_reached(limits)

Check if any of limits reached.

.. versionadded:: 0.8.0

Parameters

limits : Iterable of :obj:onetl.base.base_file_limit.BaseFileLimit Limits to test.

Returns

True if any of limits is reached, False otherwise.

If no limits are passed, returns False.

Examples

from onetl.file.limit import MaxFilesCount, limits_reached, limits_stop_at from onetl.impl import LocalPath limits = [MaxFilesCount(2)] limits_reached(limits) False limits_stop_at(LocalPath("/path/to/file1.csv"), limits) False limits_stop_at(LocalPath("/path/to/file2.csv"), limits) False limits_stop_at(LocalPath("/path/to/file3.csv"), limits) True limits_reached(limits) True

Source code in onetl/file/limit/limits_reached.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def limits_reached(limits: Iterable[BaseFileLimit]) -> bool:
    """
    Check if any of limits reached.

    .. versionadded:: 0.8.0

    Parameters
    ----------
    limits : Iterable of :obj:`onetl.base.base_file_limit.BaseFileLimit`
        Limits to test.

    Returns
    -------
    ``True`` if any of limits is reached, ``False`` otherwise.

    If no limits are passed, returns ``False``.

    Examples
    --------

    >>> from onetl.file.limit import MaxFilesCount, limits_reached, limits_stop_at
    >>> from onetl.impl import LocalPath
    >>> limits = [MaxFilesCount(2)]
    >>> limits_reached(limits)
    False
    >>> limits_stop_at(LocalPath("/path/to/file1.csv"), limits)
    False
    >>> limits_stop_at(LocalPath("/path/to/file2.csv"), limits)
    False
    >>> limits_stop_at(LocalPath("/path/to/file3.csv"), limits)
    True
    >>> limits_reached(limits)
    True
    """
    debug = log.isEnabledFor(logging.DEBUG)

    reached = []
    for limit in limits:
        if limit.is_reached:
            if not debug:
                # fast path
                return True

            reached.append(limit)

    if reached:
        log.debug("|FileLimit| Limits %r are reached", reached)
        return True

    return False