Skip to content

Лимит на максимальное количество файлов

Bases: BaseFileLimit, FrozenModel

Limits the total number of files handled by [file-downloader][] or [file-mover][].

All files until specified limit (including) will be downloaded/moved, but limit+1 will not.

This doesn't apply to directories.

Added in 0.8.0

Replaces deprecated onetl.core.FileLimit

Parameters

int

Maximum number of files to be handled.

Examples

Create filter which allows to download/move up to 100 files, but stops on 101:

from onetl.file.limit import MaxFilesCount

limit = MaxFilesCount(100)
Source code in onetl/file/limit/max_files_count.py 
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
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
class MaxFilesCount(BaseFileLimit, FrozenModel):
    """Limits the total number of files handled by [file-downloader][] or [file-mover][].

    All files until specified limit (including) will be downloaded/moved, but `limit+1` will not.

    This doesn't apply to directories.

    !!! success "Added in 0.8.0"
        Replaces deprecated `onetl.core.FileLimit`

    Parameters
    ----------

    limit : int
        Maximum number of files to be handled.

    Examples
    --------

    Create filter which allows to download/move up to 100 files, but stops on 101:

    ```python
    from onetl.file.limit import MaxFilesCount

    limit = MaxFilesCount(100)

    ```
    """

    limit: int

    _handled: int = 0

    def __init__(self, limit: int):
        # this is only to allow passing glob as positional argument
        super().__init__(limit=limit)

    def __repr__(self):
        return f"{self.__class__.__name__}({self.limit})"

    @validator("limit")
    def _limit_cannot_be_negative(cls, value):
        if value <= 0:
            msg = "Limit should be positive number"
            raise ValueError(msg)
        return value

    def reset(self):
        self._handled = 0
        return self

    def stops_at(self, path: PathProtocol) -> bool:
        if self.is_reached:
            return True

        if path.is_dir():
            # directories count does not matter
            return False

        self._handled += 1
        return self.is_reached

    @property
    def is_reached(self) -> bool:
        return self._handled > self.limit

is_reached property

reset()

Source code in onetl/file/limit/max_files_count.py 
65
66
67
def reset(self):
    self._handled = 0
    return self

stops_at(path)

Source code in onetl/file/limit/max_files_count.py 
69
70
71
72
73
74
75
76
77
78
def stops_at(self, path: PathProtocol) -> bool:
    if self.is_reached:
        return True

    if path.is_dir():
        # directories count does not matter
        return False

    self._handled += 1
    return self.is_reached