Skip to content

qass.tools.analyzer.deleting_process

Classes:

Name Description
DeleteHandler

DeleteHandler Class with different functionalities to delete local files on disk.

DeleteHandler

DeleteHandler Class with different functionalities to delete local files on disk.

With an instance of this class a specific deleting-pattern and path (for the files) will be associated. The programm always looks for the oldest versions of files which match with the pattern. Files will be deletet until the maximum amount of allowed files is reached or enough disk space is free. This class stands for now on its own. With a def main() syntacs it will be executed by programm start.

Parameters:

Name Type Description Default
path str

Local directory path.

 required
pattern str

Searched pattern.

 required
pattern bool

Flag if log file should be created (by default: False)

 required

Methods:

Name Description
__init__

Constructor to connect a specific local directory path with a specific pattern parsed.
delete_by_amount

Provided function to delete files based on file amount in location if amount overruns defined limit.
delete_by_disk_space

Provided method to delete files by a given maximum disk space usage.

Attributes:

Name Type Description
file_logger
full_path
log_entries
path
pattern
Source code in src/qass/tools/analyzer/deleting_process.py 
 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
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
class DeleteHandler:
    """DeleteHandler Class with different functionalities to delete local files on disk.

    With an instance of this class a specific deleting-pattern and path (for the files) will be associated. The programm always looks for the oldest versions of files which match with the pattern.
    Files will be deletet until the maximum amount of allowed files is reached or enough disk space is free. This class stands for now on its own. With a def main() syntacs it will be executed by programm start.

    Parameters
    ----------
    path : str
        Local directory path.
    pattern : str
        Searched pattern.
    pattern : bool, optional
        Flag if log file should be created (by default: False)
    """

    def __init__(self, path: str, pattern: str, log_entries=False) -> None:
        """Constructor to connect a specific local directory path with a specific pattern parsed.

        Provided functions can be applied later for each combination. Pattern has to be according to the rules used by Unix shell (used glob module is based on that). In the following Unix Shell rules are provided:

        | Pattern | Meaning                               | Example                                                                              |
        | :------ | :----------------------------------- | :------------------------------------------------------------------------------------ |
        | *	      | Matches everything	                  | *.pdf matches all files with the pdf extension                                       |
        | ?	      | Matches any single character	      | sales/??.jpeg matches all files with two characters long present in the sales folder |
        | []	  | Matches any character in the sequence |	[psr]* matches all files starting with the letter p, s, or r                         |
        | [!]	  | Matches any character not in sequence |	[!psr]* matches all files not starting with the letter p, s, or r                    |

        For a literal match, wrap the meta-characters in brackets. For example, '[?]' matches the character '?'

        Note
        ----
        Created logfile be ignored automatically.

        Parameters
        ----------
        path : str
            Local directory path.
        pattern : str
            Searched pattern.
        """
        self.path = path
        self.pattern = pattern
        self.log_entries = log_entries
        self.full_path = str(Path(self.path) / Path(self.pattern))
        if self.log_entries:
            # logfilesize_limit = amount of entires * average entry size(=200bytes)
            logfilesize_limit = 10000 * 200
            self.file_logger = self._create_file_logger_obj(
                self.path, self.pattern, logfilesize_limit
            )

    def delete_by_amount(self, max_amount: int) -> None:
        """Provided function to delete files based on file amount in location if amount overruns defined limit.

        Parameters
        ----------
        max_amount : int
            Maximum amount (limit) of files allowed in this directory which are match pattern.
        """
        # search for files in directory which match pattern
        delete_list = glob.glob(self.full_path)  # delete_list is saved with full path
        file_amount = len(delete_list)
        if file_amount <= 0:
            if self.log_entries:
                self.file_logger.warning(
                    "The current deleting directory has no matching files to delete. Check your pattern or check if the directory is correct."
                )
            return
        # calc how many files have to be deletet in order to satisfy limit
        amount_to_delete = file_amount - max_amount

        # if there are any files to delete:
        if amount_to_delete > 0:
            # sort list for oldest files
            sorted_list = self.__get_oldest(delete_list)
            for idx in range(0, amount_to_delete):
                self.__delete_file(sorted_list[idx][1])

    def delete_by_disk_space(self, disk_usage_limit: float) -> None:
        """Provided method to delete files by a given maximum disk space usage.

        Parameters
        ----------
        disk_usage_limit : float
            Limit how much disk memory is allowed to use at maximum. Parsed as part of the whole.

        Raises
        ------
        OSError
            Exception is raised if deleting process cannot be compelted.
            Either because there are not enough files that match pattern to satisfy limit or because any other wild error appears.
        """
        # read out currently used disk usage
        disk_usage = shutil.disk_usage(self.path)
        # help to debug: disk_usage[0] = total // disk_usage[1] = used // disk_usage[2] = free
        # calc free space that is required based on user limit
        target_free_space = disk_usage[0] * (1 - disk_usage_limit)
        # calc therefore required space
        space_to_make = abs(target_free_space - disk_usage[2])
        if space_to_make > 0:
            delete_list = glob.glob(
                self.full_path
            )  # delete_list is saved with full path
            if len(delete_list) == 0:
                if self.log_entries:
                    self.file_logger.warning(
                        "The current deleting directory has no matching files to delete. Check your pattern or check if the directory is correct."
                    )
                return
            # sort list for oldest files
            sorted_list = self.__get_oldest(delete_list)
            # helper
            filesize = 0
            # n = -1
            for n, (date, file) in enumerate(sorted_list):
                filesize = filesize + os.path.getsize(file)
                if filesize >= space_to_make:
                    break
            for idx in range(0, n + 1):
                self.__delete_file(sorted_list[idx][1])

    def __get_oldest(self, possible_files: List) -> List:
        """Private method to get a list of local files which are sorted by creation date.

        Parameters
        ----------
        possible_files : list
            Everything in directory which matches pattern.

        Returns
        -------
        list
            Sorted list with the oldest files as first entry.
        """
        # helper
        creation_dates = []
        for file in possible_files:
            if str(file).endswith(".log"):
                continue
            # read out creation date (here: creation_time) of each file
            creation_time = os.path.getctime(file)
            # convert to easy readable time
            creation_time = time.ctime(creation_time)
            # append to list
            creation_dates.append(creation_time)
        # zip file name and creation time
        ziped_lists = zip(creation_dates, possible_files)
        # sort for oldest
        sorted_ziped_lists = sorted(ziped_lists, key=lambda x: x[0])
        return sorted_ziped_lists

    def __delete_file(self, deleting_file: str) -> None:
        """Private method to delete parsed file.

        Parameters
        ----------
        deleting_file: str
            File(path) that should be deletet.

        Raises
        ------
        OSError
            An error is raisen if files cannot be removed (arbitrary reasons).
        """
        try:
            os.remove(deleting_file)
            if self.log_entries:
                self.file_logger.info(f"File: {deleting_file} removed sucessfull")
        except OSError as error:
            if self.log_entries:
                self.file_logger.error(error)
                self.file_logger.error(f"File {deleting_file} could not be removed.")

    def _create_file_logger_obj(
        self,
        path: Union[str, Path, os.PathLike],
        pattern_to_log: str,
        filesize_limit: int,
    ):
        """Creates formatete object from python built in logging module.

        Creates a RotatingFileHandler object which logs all occuring events. Logged will be time, name of function and
        message itself. The logfile name will be created automatically by used search pattern and stored in supervised
        directory. If filesize limit is reached a new logfile will be created.
        Additionally, all occuring error loggings will be send to sys.stderr.

        Note
        ----
        In Analyzer environment all message going to sys-stderr will be stored in internal log.

        Parameters
        ----------
        pattern_to_log : str
            Path of supervised directory
        pattern_to_log : str
            Pattern which is used in constrcutor of DeleteHandler class
        filesize_limit : int
            Maximum logfile size in bytes

        Returns
        -------
        logging.handlers.RotatingFileHandler
            logger object
        """
        # create logger
        logger_name = pattern_to_log + "_file_logger"
        logger_obj = logging.getLogger(logger_name)

        file_path = Path(path) / Path(pattern_to_log + ".log")
        # define logger
        # set logging level to lowest setting
        logger_obj.setLevel(logging.DEBUG)
        # create rotating file handler and set level to debug
        rfh = logging.handlers.RotatingFileHandler(
            str(file_path), maxBytes=filesize_limit, backupCount=1
        )
        rfh.setLevel(logging.DEBUG)
        # create sys.stderr handler
        ch_stderr = logging.StreamHandler(sys.stderr)
        ch_stderr.setLevel(logging.ERROR)
        # create formatter
        formatter = logging.Formatter("[%(asctime)s]:  %(levelname)s - %(message)s")
        # add formatter to rfh and ch_stderr
        rfh.setFormatter(formatter)
        ch_stderr.setFormatter(formatter)
        # add ch to logger
        logger_obj.addHandler(rfh)
        logger_obj.addHandler(ch_stderr)

        return logger_obj

file_logger instance-attribute 

file_logger = _create_file_logger_obj(path, pattern, logfilesize_limit)

full_path instance-attribute 

full_path = str(Path(path) / Path(pattern))

log_entries instance-attribute 

log_entries = log_entries

path instance-attribute 

path = path

pattern instance-attribute 

pattern = pattern

__delete_file 

__delete_file(deleting_file: str) -> None

Private method to delete parsed file.

Parameters:

Name Type Description Default
deleting_file str

File(path) that should be deletet.

 required

Raises:

Type Description
OSError

An error is raisen if files cannot be removed (arbitrary reasons).
Source code in src/qass/tools/analyzer/deleting_process.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
def __delete_file(self, deleting_file: str) -> None:
    """Private method to delete parsed file.

    Parameters
    ----------
    deleting_file: str
        File(path) that should be deletet.

    Raises
    ------
    OSError
        An error is raisen if files cannot be removed (arbitrary reasons).
    """
    try:
        os.remove(deleting_file)
        if self.log_entries:
            self.file_logger.info(f"File: {deleting_file} removed sucessfull")
    except OSError as error:
        if self.log_entries:
            self.file_logger.error(error)
            self.file_logger.error(f"File {deleting_file} could not be removed.")

__get_oldest 

__get_oldest(possible_files: List) -> List

Private method to get a list of local files which are sorted by creation date.

Parameters:

Name Type Description Default
possible_files list

Everything in directory which matches pattern.

 required

Returns:

Type Description
list

Sorted list with the oldest files as first entry.
Source code in src/qass/tools/analyzer/deleting_process.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def __get_oldest(self, possible_files: List) -> List:
    """Private method to get a list of local files which are sorted by creation date.

    Parameters
    ----------
    possible_files : list
        Everything in directory which matches pattern.

    Returns
    -------
    list
        Sorted list with the oldest files as first entry.
    """
    # helper
    creation_dates = []
    for file in possible_files:
        if str(file).endswith(".log"):
            continue
        # read out creation date (here: creation_time) of each file
        creation_time = os.path.getctime(file)
        # convert to easy readable time
        creation_time = time.ctime(creation_time)
        # append to list
        creation_dates.append(creation_time)
    # zip file name and creation time
    ziped_lists = zip(creation_dates, possible_files)
    # sort for oldest
    sorted_ziped_lists = sorted(ziped_lists, key=lambda x: x[0])
    return sorted_ziped_lists

__init__ 

__init__(path: str, pattern: str, log_entries=False) -> None

Constructor to connect a specific local directory path with a specific pattern parsed.

Provided functions can be applied later for each combination. Pattern has to be according to the rules used by Unix shell (used glob module is based on that). In the following Unix Shell rules are provided:

Pattern Meaning Example
* Matches everything *.pdf matches all files with the pdf extension
? Matches any single character sales/??.jpeg matches all files with two characters long present in the sales folder
[] Matches any character in the sequence [psr]* matches all files starting with the letter p, s, or r
[!] Matches any character not in sequence [!psr]* matches all files not starting with the letter p, s, or r

For a literal match, wrap the meta-characters in brackets. For example, '[?]' matches the character '?'

Note

Created logfile be ignored automatically.

Parameters:

Name Type Description Default
path str

Local directory path.

 required
pattern str

Searched pattern.

 required
Source code in src/qass/tools/analyzer/deleting_process.py 
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
83
def __init__(self, path: str, pattern: str, log_entries=False) -> None:
    """Constructor to connect a specific local directory path with a specific pattern parsed.

    Provided functions can be applied later for each combination. Pattern has to be according to the rules used by Unix shell (used glob module is based on that). In the following Unix Shell rules are provided:

    | Pattern | Meaning                               | Example                                                                              |
    | :------ | :----------------------------------- | :------------------------------------------------------------------------------------ |
    | *	      | Matches everything	                  | *.pdf matches all files with the pdf extension                                       |
    | ?	      | Matches any single character	      | sales/??.jpeg matches all files with two characters long present in the sales folder |
    | []	  | Matches any character in the sequence |	[psr]* matches all files starting with the letter p, s, or r                         |
    | [!]	  | Matches any character not in sequence |	[!psr]* matches all files not starting with the letter p, s, or r                    |

    For a literal match, wrap the meta-characters in brackets. For example, '[?]' matches the character '?'

    Note
    ----
    Created logfile be ignored automatically.

    Parameters
    ----------
    path : str
        Local directory path.
    pattern : str
        Searched pattern.
    """
    self.path = path
    self.pattern = pattern
    self.log_entries = log_entries
    self.full_path = str(Path(self.path) / Path(self.pattern))
    if self.log_entries:
        # logfilesize_limit = amount of entires * average entry size(=200bytes)
        logfilesize_limit = 10000 * 200
        self.file_logger = self._create_file_logger_obj(
            self.path, self.pattern, logfilesize_limit
        )

delete_by_amount 

delete_by_amount(max_amount: int) -> None

Provided function to delete files based on file amount in location if amount overruns defined limit.

Parameters:

Name Type Description Default
max_amount int

Maximum amount (limit) of files allowed in this directory which are match pattern.

 required
Source code in src/qass/tools/analyzer/deleting_process.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def delete_by_amount(self, max_amount: int) -> None:
    """Provided function to delete files based on file amount in location if amount overruns defined limit.

    Parameters
    ----------
    max_amount : int
        Maximum amount (limit) of files allowed in this directory which are match pattern.
    """
    # search for files in directory which match pattern
    delete_list = glob.glob(self.full_path)  # delete_list is saved with full path
    file_amount = len(delete_list)
    if file_amount <= 0:
        if self.log_entries:
            self.file_logger.warning(
                "The current deleting directory has no matching files to delete. Check your pattern or check if the directory is correct."
            )
        return
    # calc how many files have to be deletet in order to satisfy limit
    amount_to_delete = file_amount - max_amount

    # if there are any files to delete:
    if amount_to_delete > 0:
        # sort list for oldest files
        sorted_list = self.__get_oldest(delete_list)
        for idx in range(0, amount_to_delete):
            self.__delete_file(sorted_list[idx][1])

delete_by_disk_space 

delete_by_disk_space(disk_usage_limit: float) -> None

Provided method to delete files by a given maximum disk space usage.

Parameters:

Name Type Description Default
disk_usage_limit float

Limit how much disk memory is allowed to use at maximum. Parsed as part of the whole.

 required

Raises:

Type Description
OSError

Exception is raised if deleting process cannot be compelted. Either because there are not enough files that match pattern to satisfy limit or because any other wild error appears.
Source code in src/qass/tools/analyzer/deleting_process.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
def delete_by_disk_space(self, disk_usage_limit: float) -> None:
    """Provided method to delete files by a given maximum disk space usage.

    Parameters
    ----------
    disk_usage_limit : float
        Limit how much disk memory is allowed to use at maximum. Parsed as part of the whole.

    Raises
    ------
    OSError
        Exception is raised if deleting process cannot be compelted.
        Either because there are not enough files that match pattern to satisfy limit or because any other wild error appears.
    """
    # read out currently used disk usage
    disk_usage = shutil.disk_usage(self.path)
    # help to debug: disk_usage[0] = total // disk_usage[1] = used // disk_usage[2] = free
    # calc free space that is required based on user limit
    target_free_space = disk_usage[0] * (1 - disk_usage_limit)
    # calc therefore required space
    space_to_make = abs(target_free_space - disk_usage[2])
    if space_to_make > 0:
        delete_list = glob.glob(
            self.full_path
        )  # delete_list is saved with full path
        if len(delete_list) == 0:
            if self.log_entries:
                self.file_logger.warning(
                    "The current deleting directory has no matching files to delete. Check your pattern or check if the directory is correct."
                )
            return
        # sort list for oldest files
        sorted_list = self.__get_oldest(delete_list)
        # helper
        filesize = 0
        # n = -1
        for n, (date, file) in enumerate(sorted_list):
            filesize = filesize + os.path.getsize(file)
            if filesize >= space_to_make:
                break
        for idx in range(0, n + 1):
            self.__delete_file(sorted_list[idx][1])