Skip to content

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

limits_reached(limits)

Check if any of limits reached.

Added in 0.8.0

Parameters

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

Returns

bool 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 
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
63
64
65
66
67
def limits_reached(limits: Iterable[BaseFileLimit]) -> bool:
    """
    Check if any of limits reached.

    !!! success "Added in 0.8.0"

    Parameters
    ----------
    limits : Iterable of [onetl.base.base_file_limit.BaseFileLimit][]
        Limits to test.

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

        If no limits are passed, returns `False`.

    Examples
    --------

    ```python
    >>> 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