Skip to content

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

Bases: BaseFileLimit, FrozenModel

Limits the total number of files handled by :ref:file-downloader or :ref:file-mover.

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

This doesn't apply to directories.

.. versionadded:: 0.8.0 Replaces deprecated onetl.core.FileLimit

Parameters

limit : int

Examples

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

.. code:: python

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
class MaxFilesCount(BaseFileLimit, FrozenModel):
    """Limits the total number of files handled by :ref:`file-downloader` or :ref:`file-mover`.

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

    This doesn't apply to directories.

    .. versionadded:: 0.8.0
        Replaces deprecated ``onetl.core.FileLimit``

    Parameters
    ----------

    limit : int

    Examples
    --------

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

    .. code:: 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)  # type: ignore

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

    @validator("limit")
    def _limit_cannot_be_negative(cls, value):
        if value <= 0:
            raise ValueError("Limit should be positive number")
        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 
62
63
64
def reset(self):
    self._handled = 0
    return self

stops_at(path)

Source code in onetl/file/limit/max_files_count.py 
66
67
68
69
70
71
72
73
74
75
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