limits_reached

onetl.file.limit.limits_reached.limits_reached(limits: Iterable[BaseFileLimit]) → bool

Check if any of limits reached.

Parameters:

limitsIterable of 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)]
assert not limits_reached(limits)

assert not limits_stop_at(LocalPath("/path/to/file.csv"), limits)
assert limits_stop_at(LocalPath("/path/to/file.csv"), limits)

assert limits_reached(limits)