TotalFilesSize

Bases: BaseFileLimit, FrozenModel

Limits the total size of files handled by File Downloader or File Mover.

Calculates the sum of downloaded/moved files size (.stat().st_size), and checks that this sum is less or equal to specified limit.

After limit is reached, no more files will be downloaded/moved.

Doesn't affect directories, paths without .stat() method or files with zero size.

Added in 0.13.0

Note

SI unit prefixes means that 1KB == 1 kilobyte == 1000 bytes. If you need 1024 bytes, use 1 KiB == 1 kibibyte.

Parameters:

• limit (int | str) –

  Maximum total size of files to handle. Accepts bytes as int or human-readable string, e.g. "1GiB".

Examples:

Create filter which allows to download/move files with total size up to 1GiB, but not higher:

from onetl.file.limit import MaxFilesCount

limit = TotalFilesSize("1GiB")

is_reached property

Check if limit is reached.

Added in 0.8.0

Returns:

• bool –

  True if limit is reached, False otherwise.

Examples:

>>> from onetl.impl import LocalPath
>>> limit.is_reached
False
>>> limit.stops_at(LocalPath("/path/to/file.csv"))
False
>>> limit.is_reached
False
>>> # after limit is reached
>>> limit.stops_at(LocalPath("/path/to/file.csv"))
True
>>> limit.is_reached
True

reset()

Resets the internal limit state.

Added in 0.8.0

Returns:

• Self –

  Returns a filter of the same type, but with non-reached state.

  It could be the same filter or a new one, this is an implementation detail.

Examples:

>>> limit.is_reached
True
>>> new_limit = limit.reset()
>>> new_limit.is_reached
False

stops_at(path)

Update internal state and return current state.

Added in 0.8.0

Parameters:

• path (PathProtocol) –

  Path to check

Returns:

• bool –

  True if limit is reached, False otherwise.

Examples:

>>> from onetl.impl import LocalPath
>>> # limit is not reached yet
>>> limit.stops_at(LocalPath("/path/to/file.csv"))
False
>>> # after limit is reached
>>> limit.stops_at(LocalPath("/path/to/another.csv"))
True
>>> # at this point, .stops_at() and .is_reached will always return True,
>>> # even on inputs that returned False before.
>>> # it will be in the same state until .reset() is called
>>> limit.stops_at(LocalPath("/path/to/file.csv"))
True