Skip to content

pyExpireBackups API Documentation

expire

Created on 2022-04-01

@author: wf

BackupFile

a Backup file which is potentially to be expired

Source code in expirebackups/expire.py
 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
 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
class BackupFile:
    """
    a Backup file which is potentially to be expired
    """

    def __init__(self, filePath: str):
        """
        constructor

        Args:
            filePath(str): the filePath of this backup File
        """
        self.filePath = filePath
        self.modified, self.size = self.getStats()
        self.sizeValue, self.unit, self.factor = BackupFile.getSize(self.size)
        self.sizeString = BackupFile.getSizeString(self.size)
        self.ageInDays = self.getAgeInDays()
        self.isoDate = self.getIsoDateOfModification()
        self.expire = False

    def __str__(self):
        """
        return a string representation of me
        """
        text = f"{self.ageInDays:6.1f} days {self.getMarker()}({self.sizeString}):{self.filePath}"
        return text

    def getMarker(self):
        """
        get my marker

        Returns:
            str: a symbol ❌ if i am to be deleted a ✅ if i am going to be kept
        """
        marker = "❌" if self.expire else "✅"
        return marker

    @classmethod
    def getSizeString(cls, size: float) -> str:
        """
        get my Size in human readable terms as a s

        Args:
            size(float): Size in Bytes

        Returns:
            str: a String representation
        """
        size, unit, _factor = cls.getSize(size)
        text = f"{size:5.0f} {unit}"
        return text

    @classmethod
    def getSize(cls, size: float) -> Tuple[float, str, float]:
        """
        get my Size in human readable terms

        Args:
            size(float): Size in Bytes

        Returns:
            Tuple(float,str,float): the size, unit and factor of the unit e.g. 3.2, "KB", 1024
        """
        units = [" B", "KB", "MB", "GB", "TB"]
        unitIndex = 0
        factor = 1
        while size > 1024:
            factor = factor * 1024
            size = size / 1024
            unitIndex += 1
        return size, units[unitIndex], factor

    def getStats(self) -> Tuple[datetime.datetime, float]:
        """
        get the datetime when the file was modified

        Returns:
            datetime: the file modification time
        """
        stats = os.stat(self.filePath)
        modified = datetime.datetime.fromtimestamp(stats.st_mtime, tz=datetime.timezone.utc)
        size = stats.st_size
        return modified, size

    def getAgeInDays(self) -> float:
        """
        get the age of this backup file in days

        Returns:
            float: the number of days this file is old
        """
        now = datetime.datetime.now(tz=datetime.timezone.utc)
        age = now - self.modified
        return age.days

    def getIsoDateOfModification(self):
        """
        get the data of modification as an ISO date string

        Returns:
            str: an iso representation of the modification date
        """
        isoDate = self.modified.strftime("%Y-%m-%d_%H:%M")
        return isoDate

    def delete(self):
        """
        delete my file
        """
        if os.path.isfile(self.filePath):
            os.remove(self.filePath)

__init__(filePath)

constructor

Parameters:

Name Type Description Default
filePath(str)

the filePath of this backup File

required
Source code in expirebackups/expire.py
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def __init__(self, filePath: str):
    """
    constructor

    Args:
        filePath(str): the filePath of this backup File
    """
    self.filePath = filePath
    self.modified, self.size = self.getStats()
    self.sizeValue, self.unit, self.factor = BackupFile.getSize(self.size)
    self.sizeString = BackupFile.getSizeString(self.size)
    self.ageInDays = self.getAgeInDays()
    self.isoDate = self.getIsoDateOfModification()
    self.expire = False

__str__()

return a string representation of me

Source code in expirebackups/expire.py
50
51
52
53
54
55
def __str__(self):
    """
    return a string representation of me
    """
    text = f"{self.ageInDays:6.1f} days {self.getMarker()}({self.sizeString}):{self.filePath}"
    return text

delete()

delete my file

Source code in expirebackups/expire.py
135
136
137
138
139
140
def delete(self):
    """
    delete my file
    """
    if os.path.isfile(self.filePath):
        os.remove(self.filePath)

getAgeInDays()

get the age of this backup file in days

Returns:

Name Type Description
float float

the number of days this file is old

Source code in expirebackups/expire.py
114
115
116
117
118
119
120
121
122
123
def getAgeInDays(self) -> float:
    """
    get the age of this backup file in days

    Returns:
        float: the number of days this file is old
    """
    now = datetime.datetime.now(tz=datetime.timezone.utc)
    age = now - self.modified
    return age.days

getIsoDateOfModification()

get the data of modification as an ISO date string

Returns:

Name Type Description
str

an iso representation of the modification date

Source code in expirebackups/expire.py
125
126
127
128
129
130
131
132
133
def getIsoDateOfModification(self):
    """
    get the data of modification as an ISO date string

    Returns:
        str: an iso representation of the modification date
    """
    isoDate = self.modified.strftime("%Y-%m-%d_%H:%M")
    return isoDate

getMarker()

get my marker

Returns:

Name Type Description
str

a symbol ❌ if i am to be deleted a ✅ if i am going to be kept

Source code in expirebackups/expire.py
57
58
59
60
61
62
63
64
65
def getMarker(self):
    """
    get my marker

    Returns:
        str: a symbol ❌ if i am to be deleted a ✅ if i am going to be kept
    """
    marker = "❌" if self.expire else "✅"
    return marker

getSize(size) classmethod

get my Size in human readable terms

Parameters:

Name Type Description Default
size(float)

Size in Bytes

required

Returns:

Name Type Description
Tuple (float, str, float)

the size, unit and factor of the unit e.g. 3.2, "KB", 1024

Source code in expirebackups/expire.py
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
@classmethod
def getSize(cls, size: float) -> Tuple[float, str, float]:
    """
    get my Size in human readable terms

    Args:
        size(float): Size in Bytes

    Returns:
        Tuple(float,str,float): the size, unit and factor of the unit e.g. 3.2, "KB", 1024
    """
    units = [" B", "KB", "MB", "GB", "TB"]
    unitIndex = 0
    factor = 1
    while size > 1024:
        factor = factor * 1024
        size = size / 1024
        unitIndex += 1
    return size, units[unitIndex], factor

getSizeString(size) classmethod

get my Size in human readable terms as a s

Parameters:

Name Type Description Default
size(float)

Size in Bytes

required

Returns:

Name Type Description
str str

a String representation

Source code in expirebackups/expire.py
67
68
69
70
71
72
73
74
75
76
77
78
79
80
@classmethod
def getSizeString(cls, size: float) -> str:
    """
    get my Size in human readable terms as a s

    Args:
        size(float): Size in Bytes

    Returns:
        str: a String representation
    """
    size, unit, _factor = cls.getSize(size)
    text = f"{size:5.0f} {unit}"
    return text

getStats()

get the datetime when the file was modified

Returns:

Name Type Description
datetime Tuple[datetime, float]

the file modification time

Source code in expirebackups/expire.py
102
103
104
105
106
107
108
109
110
111
112
def getStats(self) -> Tuple[datetime.datetime, float]:
    """
    get the datetime when the file was modified

    Returns:
        datetime: the file modification time
    """
    stats = os.stat(self.filePath)
    modified = datetime.datetime.fromtimestamp(stats.st_mtime, tz=datetime.timezone.utc)
    size = stats.st_size
    return modified, size

Expiration

Expiration pattern

Source code in expirebackups/expire.py
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
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
class Expiration:
    """
    Expiration pattern
    """

    def __init__(
        self,
        days: int = defaultDays,
        weeks: int = defaultWeeks,
        months: int = defaultMonths,
        years: int = defaultYears,
        minFileSize: int = defaultMinFileSize,
        debug: bool = False,
    ):
        """
        constructor

        Args:
            days(float): how many files to keep for the daily backup
            weeks(float): how many files to keep for the weekly backup
            months(float): how many files to keep for the monthly backup
            years(float):  how many files to keep for the yearly backup
            debug(bool): if true show debug information (rule application)
        """
        self.rules = {
            "dayly": ExpirationRule("days", 1.0, days),
            "weekly": ExpirationRule("weeks", 7.0, weeks),
            # the month is in fact 4 weeks
            "monthly": ExpirationRule("months", 28.0, months),
            # the year is in fact 52 weeks or 13 of the 4 week months
            "yearly": ExpirationRule("years", 364.0, years),
        }
        self.minFileSize = minFileSize
        self.debug = debug

    def getNextRule(self, ruleIter, prevFile: BackupFile, verbose: bool) -> ExpirationRule:
        """
        get the next rule for the given ruleIterator

        Args:
            ruleIter(Iter): Iterator over ExpirationRules
            prevFile(BackupFile): the previousFile to take into account / reset/anchor the rule with
            verbose(bool): if True show a message that the rule will be applied
        Returns:
            ExpirationRule: the next ExpirationRule
        """
        ruleKey = next(ruleIter)
        rule = self.rules[ruleKey]
        rule.ruleName = ruleKey
        if verbose:
            print(f"keeping {rule.minAmount} files for {rule.ruleName} backup")
        rule.reset(prevFile)
        return rule

    def applyRules(self, backupFiles: list, verbose: bool = True):
        """
        apply my expiration rules to the given list of
        backup Files

        Args:
            backupFiles(list): the list of backupFiles to apply the rules to
            verbose(debug): if true show what the rules are doing
        Returns:
            list: the sorted and marked list of backupFiles
        """
        filesByAge = sorted(backupFiles, key=lambda backupFile: backupFile.getAgeInDays())
        ruleIter = iter(self.rules)
        rule = self.getNextRule(ruleIter, None, verbose)
        prevFile = None
        for file in filesByAge:
            if file.size < self.minFileSize:
                file.expire = True
            else:
                ruleDone = rule.apply(file, prevFile, debug=self.debug)
                if not file.expire:
                    prevFile = file
                if ruleDone:
                    rule = self.getNextRule(ruleIter, prevFile, verbose)
        return filesByAge

__init__(days=defaultDays, weeks=defaultWeeks, months=defaultMonths, years=defaultYears, minFileSize=defaultMinFileSize, debug=False)

constructor

Parameters:

Name Type Description Default
days(float)

how many files to keep for the daily backup

required
weeks(float)

how many files to keep for the weekly backup

required
months(float)

how many files to keep for the monthly backup

required
years(float)

how many files to keep for the yearly backup

required
debug(bool)

if true show debug information (rule application)

required
Source code in expirebackups/expire.py
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
def __init__(
    self,
    days: int = defaultDays,
    weeks: int = defaultWeeks,
    months: int = defaultMonths,
    years: int = defaultYears,
    minFileSize: int = defaultMinFileSize,
    debug: bool = False,
):
    """
    constructor

    Args:
        days(float): how many files to keep for the daily backup
        weeks(float): how many files to keep for the weekly backup
        months(float): how many files to keep for the monthly backup
        years(float):  how many files to keep for the yearly backup
        debug(bool): if true show debug information (rule application)
    """
    self.rules = {
        "dayly": ExpirationRule("days", 1.0, days),
        "weekly": ExpirationRule("weeks", 7.0, weeks),
        # the month is in fact 4 weeks
        "monthly": ExpirationRule("months", 28.0, months),
        # the year is in fact 52 weeks or 13 of the 4 week months
        "yearly": ExpirationRule("years", 364.0, years),
    }
    self.minFileSize = minFileSize
    self.debug = debug

applyRules(backupFiles, verbose=True)

apply my expiration rules to the given list of backup Files

Parameters:

Name Type Description Default
backupFiles(list)

the list of backupFiles to apply the rules to

required
verbose(debug)

if true show what the rules are doing

required

Returns: list: the sorted and marked list of backupFiles

Source code in expirebackups/expire.py
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
def applyRules(self, backupFiles: list, verbose: bool = True):
    """
    apply my expiration rules to the given list of
    backup Files

    Args:
        backupFiles(list): the list of backupFiles to apply the rules to
        verbose(debug): if true show what the rules are doing
    Returns:
        list: the sorted and marked list of backupFiles
    """
    filesByAge = sorted(backupFiles, key=lambda backupFile: backupFile.getAgeInDays())
    ruleIter = iter(self.rules)
    rule = self.getNextRule(ruleIter, None, verbose)
    prevFile = None
    for file in filesByAge:
        if file.size < self.minFileSize:
            file.expire = True
        else:
            ruleDone = rule.apply(file, prevFile, debug=self.debug)
            if not file.expire:
                prevFile = file
            if ruleDone:
                rule = self.getNextRule(ruleIter, prevFile, verbose)
    return filesByAge

getNextRule(ruleIter, prevFile, verbose)

get the next rule for the given ruleIterator

Parameters:

Name Type Description Default
ruleIter(Iter)

Iterator over ExpirationRules

required
prevFile(BackupFile)

the previousFile to take into account / reset/anchor the rule with

required
verbose(bool)

if True show a message that the rule will be applied

required

Returns: ExpirationRule: the next ExpirationRule

Source code in expirebackups/expire.py
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
def getNextRule(self, ruleIter, prevFile: BackupFile, verbose: bool) -> ExpirationRule:
    """
    get the next rule for the given ruleIterator

    Args:
        ruleIter(Iter): Iterator over ExpirationRules
        prevFile(BackupFile): the previousFile to take into account / reset/anchor the rule with
        verbose(bool): if True show a message that the rule will be applied
    Returns:
        ExpirationRule: the next ExpirationRule
    """
    ruleKey = next(ruleIter)
    rule = self.rules[ruleKey]
    rule.ruleName = ruleKey
    if verbose:
        print(f"keeping {rule.minAmount} files for {rule.ruleName} backup")
    rule.reset(prevFile)
    return rule

ExpirationRule

an expiration rule keeps files at a certain

Source code in expirebackups/expire.py
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
class ExpirationRule:
    """
    an expiration rule keeps files at a certain
    """

    def __init__(self, name, freq: float, minAmount: int):
        """
        constructor

        name(str): name of this rule
        freq(float): the frequency) in days
        minAmount(int): the minimum of files to keep around
        """
        self.name = name
        self.ruleName = name  # will late be changed by a sideEffect in getNextRule e.g. from "week" to "weekly"
        self.freq = freq
        self.minAmount = minAmount
        if minAmount < 0:
            raise Exception(f"{self.minAmount} {self.name} is invalid - {self.name} must be >=0")

    def reset(self, prevFile: BackupFile):
        """
        reset my state with the given previous File

        Args:
            prevFile: BackupFile - the file to anchor my startAge with
        """
        self.kept = 0
        if prevFile is None:
            self.startAge = 0
        else:
            self.startAge = prevFile.ageInDays

    def apply(self, file: BackupFile, prevFile: BackupFile, debug: bool) -> bool:
        """
        apply me to the given file taking the previously kept File prevFile (which might be None) into account

        Args:

            file(BackupFile): the file to apply this rule for
            prevFile(BackupFile): the previous file to potentially take into account
            debug(bool): if True show debug output
        """
        if prevFile is not None:
            ageDiff = file.ageInDays - prevFile.ageInDays
            keep = ageDiff >= self.freq
        else:
            ageDiff = file.ageInDays - self.startAge
            keep = True
        if keep:
            self.kept += 1
        else:
            file.expire = True
        if debug:
            print(
                f{ageDiff}({ageDiff-self.freq}) days for {self.ruleName}({self.freq}) {self.kept}/{self.minAmount}{file}"
            )
        return self.kept >= self.minAmount

__init__(name, freq, minAmount)

constructor

name(str): name of this rule freq(float): the frequency) in days minAmount(int): the minimum of files to keep around

Source code in expirebackups/expire.py
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def __init__(self, name, freq: float, minAmount: int):
    """
    constructor

    name(str): name of this rule
    freq(float): the frequency) in days
    minAmount(int): the minimum of files to keep around
    """
    self.name = name
    self.ruleName = name  # will late be changed by a sideEffect in getNextRule e.g. from "week" to "weekly"
    self.freq = freq
    self.minAmount = minAmount
    if minAmount < 0:
        raise Exception(f"{self.minAmount} {self.name} is invalid - {self.name} must be >=0")

apply(file, prevFile, debug)

apply me to the given file taking the previously kept File prevFile (which might be None) into account

Args:

file(BackupFile): the file to apply this rule for
prevFile(BackupFile): the previous file to potentially take into account
debug(bool): if True show debug output
Source code in expirebackups/expire.py
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
def apply(self, file: BackupFile, prevFile: BackupFile, debug: bool) -> bool:
    """
    apply me to the given file taking the previously kept File prevFile (which might be None) into account

    Args:

        file(BackupFile): the file to apply this rule for
        prevFile(BackupFile): the previous file to potentially take into account
        debug(bool): if True show debug output
    """
    if prevFile is not None:
        ageDiff = file.ageInDays - prevFile.ageInDays
        keep = ageDiff >= self.freq
    else:
        ageDiff = file.ageInDays - self.startAge
        keep = True
    if keep:
        self.kept += 1
    else:
        file.expire = True
    if debug:
        print(
            f{ageDiff}({ageDiff-self.freq}) days for {self.ruleName}({self.freq}) {self.kept}/{self.minAmount}{file}"
        )
    return self.kept >= self.minAmount

reset(prevFile)

reset my state with the given previous File

Parameters:

Name Type Description Default
prevFile BackupFile

BackupFile - the file to anchor my startAge with

required
Source code in expirebackups/expire.py
163
164
165
166
167
168
169
170
171
172
173
174
def reset(self, prevFile: BackupFile):
    """
    reset my state with the given previous File

    Args:
        prevFile: BackupFile - the file to anchor my startAge with
    """
    self.kept = 0
    if prevFile is None:
        self.startAge = 0
    else:
        self.startAge = prevFile.ageInDays

ExpireBackups

Bases: object

Expiration of Backups - migrated from com.bitplan.backup java solution

Source code in expirebackups/expire.py
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
class ExpireBackups(object):
    """
    Expiration of Backups - migrated from com.bitplan.backup java solution
    """

    def __init__(
        self,
        rootPath: str,
        baseName: str = None,
        ext: str = None,
        expiration: Expiration = None,
        dryRun: bool = True,
        debug: bool = False,
    ):
        """
        Constructor

        Args:
            rootPath(str): the base path for this backup expiration
            baseName(str): the basename to filter for (if any)
            ext(str): file extensions to filter for e.g. ".tgz" (if any)
            expiration(Expiration): the Expiration Rules to apply
            dryRun(bool): donot delete any files but only show deletion plan
        """
        self.rootPath = rootPath
        self.baseName = baseName
        self.ext = ext
        # if no expiration is specified use the default one
        if expiration is None:
            expiration = Expiration()
        self.expiration = expiration
        self.dryRun = dryRun
        self.debug = debug

    @classmethod
    def createTestFile(cls, ageInDays: float, baseName: str = None, ext: str = ".tst"):
        """
        create a test File with the given extension and the given age in Days

        Args:
            ageInDays(float): the age of the file in days
            baseName(str): the prefix of the files (default: None)
            ext(str): the extension to be used - default ".tst"

        Returns:
            str: the full path name of the testfile
        """
        now = datetime.datetime.now(tz=datetime.timezone.utc)
        dayDelta = datetime.timedelta(days=ageInDays)
        wantedTime = now - dayDelta
        timestamp = datetime.datetime.timestamp(wantedTime)
        prefix = "" if baseName is None else f"{baseName}-"
        testFile = NamedTemporaryFile(prefix=f"{prefix}{ageInDays}daysOld-", suffix=ext, delete=False)
        with open(testFile.name, "a"):
            times = (timestamp, timestamp)  # access time and modification time
            os.utime(testFile.name, times)
        return testFile.name

    @classmethod
    def createTestFiles(cls, numberOfTestfiles: int, baseName: str = "expireBackupTest", ext: str = ".tst"):
        """
        create the given number of tests files

        Args:
            numberOfTestfiles(int): the number of files to create
            baseName(str): the prefix of the files (default: '')
            ext(str): the extension of the files (default: '.tst')

        Returns:
            tuple(str,list): the path of the directory where the test files have been created
            and a list of BackupFile files
        """
        backupFiles = []
        for ageInDays in range(1, numberOfTestfiles + 1):
            testFile = ExpireBackups.createTestFile(ageInDays, baseName=baseName, ext=ext)
            backupFiles.append(BackupFile(testFile))
        path = pathlib.Path(testFile).parent.resolve()
        return path, backupFiles

    def getBackupFiles(self) -> list:
        """
        get the list of my backup Files
        """
        backupFiles = []
        for root, _dirs, files in os.walk(self.rootPath):
            for file in files:
                include = False
                if self.baseName is not None:
                    include = file.startswith(self.baseName)
                if self.ext is not None:
                    include = file.endswith(self.ext)
                if include:
                    backupFile = BackupFile(os.path.join(root, file))
                    backupFiles.append(backupFile)
        return backupFiles

    def doexpire(self, withDelete: bool = False, show=True, showLimit: int = None):
        """
        expire the files in the given rootPath

        withDelete(bool): if True really delete the files
        show(bool): if True show the expiration plan
        showLimit(int): if set limit the number of lines to display
        """
        backupFiles = self.getBackupFiles()
        filesByAge = self.expiration.applyRules(backupFiles)
        total = 0
        keptTotal = 0
        kept = 0
        if show:
            deletehint = "by deletion" if withDelete else "dry run"
            print(f"expiring {len(filesByAge)} files {deletehint}")
        for i, backupFile in enumerate(filesByAge):
            total += backupFile.size
            totalString = BackupFile.getSizeString(total)
            marker = backupFile.getMarker()
            line = f"#{i+1:4d}{marker}:{backupFile.ageInDays:6.1f} days({backupFile.sizeString}/{totalString})→{backupFile.filePath}"
            showLine = show and showLimit is None or i < showLimit
            if showLine:
                print(line)
            if not backupFile.expire:
                kept += 1
                keptTotal += backupFile.size
            if withDelete and backupFile.expire:
                backupFile.delete()
        if show:
            keptSizeString = BackupFile.getSizeString(keptTotal)
            print(f"kept {kept} files {keptSizeString}")

__init__(rootPath, baseName=None, ext=None, expiration=None, dryRun=True, debug=False)

Constructor

Parameters:

Name Type Description Default
rootPath(str)

the base path for this backup expiration

required
baseName(str)

the basename to filter for (if any)

required
ext(str)

file extensions to filter for e.g. ".tgz" (if any)

required
expiration(Expiration)

the Expiration Rules to apply

required
dryRun(bool)

donot delete any files but only show deletion plan

required
Source code in expirebackups/expire.py
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
def __init__(
    self,
    rootPath: str,
    baseName: str = None,
    ext: str = None,
    expiration: Expiration = None,
    dryRun: bool = True,
    debug: bool = False,
):
    """
    Constructor

    Args:
        rootPath(str): the base path for this backup expiration
        baseName(str): the basename to filter for (if any)
        ext(str): file extensions to filter for e.g. ".tgz" (if any)
        expiration(Expiration): the Expiration Rules to apply
        dryRun(bool): donot delete any files but only show deletion plan
    """
    self.rootPath = rootPath
    self.baseName = baseName
    self.ext = ext
    # if no expiration is specified use the default one
    if expiration is None:
        expiration = Expiration()
    self.expiration = expiration
    self.dryRun = dryRun
    self.debug = debug

createTestFile(ageInDays, baseName=None, ext='.tst') classmethod

create a test File with the given extension and the given age in Days

Parameters:

Name Type Description Default
ageInDays(float)

the age of the file in days

required
baseName(str)

the prefix of the files (default: None)

required
ext(str)

the extension to be used - default ".tst"

required

Returns:

Name Type Description
str

the full path name of the testfile

Source code in expirebackups/expire.py
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
@classmethod
def createTestFile(cls, ageInDays: float, baseName: str = None, ext: str = ".tst"):
    """
    create a test File with the given extension and the given age in Days

    Args:
        ageInDays(float): the age of the file in days
        baseName(str): the prefix of the files (default: None)
        ext(str): the extension to be used - default ".tst"

    Returns:
        str: the full path name of the testfile
    """
    now = datetime.datetime.now(tz=datetime.timezone.utc)
    dayDelta = datetime.timedelta(days=ageInDays)
    wantedTime = now - dayDelta
    timestamp = datetime.datetime.timestamp(wantedTime)
    prefix = "" if baseName is None else f"{baseName}-"
    testFile = NamedTemporaryFile(prefix=f"{prefix}{ageInDays}daysOld-", suffix=ext, delete=False)
    with open(testFile.name, "a"):
        times = (timestamp, timestamp)  # access time and modification time
        os.utime(testFile.name, times)
    return testFile.name

createTestFiles(numberOfTestfiles, baseName='expireBackupTest', ext='.tst') classmethod

create the given number of tests files

Parameters:

Name Type Description Default
numberOfTestfiles(int)

the number of files to create

required
baseName(str)

the prefix of the files (default: '')

required
ext(str)

the extension of the files (default: '.tst')

required

Returns:

Name Type Description
tuple (str, list)

the path of the directory where the test files have been created

and a list of BackupFile files

Source code in expirebackups/expire.py
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
@classmethod
def createTestFiles(cls, numberOfTestfiles: int, baseName: str = "expireBackupTest", ext: str = ".tst"):
    """
    create the given number of tests files

    Args:
        numberOfTestfiles(int): the number of files to create
        baseName(str): the prefix of the files (default: '')
        ext(str): the extension of the files (default: '.tst')

    Returns:
        tuple(str,list): the path of the directory where the test files have been created
        and a list of BackupFile files
    """
    backupFiles = []
    for ageInDays in range(1, numberOfTestfiles + 1):
        testFile = ExpireBackups.createTestFile(ageInDays, baseName=baseName, ext=ext)
        backupFiles.append(BackupFile(testFile))
    path = pathlib.Path(testFile).parent.resolve()
    return path, backupFiles

doexpire(withDelete=False, show=True, showLimit=None)

expire the files in the given rootPath

withDelete(bool): if True really delete the files show(bool): if True show the expiration plan showLimit(int): if set limit the number of lines to display

Source code in expirebackups/expire.py
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
def doexpire(self, withDelete: bool = False, show=True, showLimit: int = None):
    """
    expire the files in the given rootPath

    withDelete(bool): if True really delete the files
    show(bool): if True show the expiration plan
    showLimit(int): if set limit the number of lines to display
    """
    backupFiles = self.getBackupFiles()
    filesByAge = self.expiration.applyRules(backupFiles)
    total = 0
    keptTotal = 0
    kept = 0
    if show:
        deletehint = "by deletion" if withDelete else "dry run"
        print(f"expiring {len(filesByAge)} files {deletehint}")
    for i, backupFile in enumerate(filesByAge):
        total += backupFile.size
        totalString = BackupFile.getSizeString(total)
        marker = backupFile.getMarker()
        line = f"#{i+1:4d}{marker}:{backupFile.ageInDays:6.1f} days({backupFile.sizeString}/{totalString})→{backupFile.filePath}"
        showLine = show and showLimit is None or i < showLimit
        if showLine:
            print(line)
        if not backupFile.expire:
            kept += 1
            keptTotal += backupFile.size
        if withDelete and backupFile.expire:
            backupFile.delete()
    if show:
        keptSizeString = BackupFile.getSizeString(keptTotal)
        print(f"kept {kept} files {keptSizeString}")

getBackupFiles()

get the list of my backup Files

Source code in expirebackups/expire.py
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
def getBackupFiles(self) -> list:
    """
    get the list of my backup Files
    """
    backupFiles = []
    for root, _dirs, files in os.walk(self.rootPath):
        for file in files:
            include = False
            if self.baseName is not None:
                include = file.startswith(self.baseName)
            if self.ext is not None:
                include = file.endswith(self.ext)
            if include:
                backupFile = BackupFile(os.path.join(root, file))
                backupFiles.append(backupFile)
    return backupFiles

main(argv=None)

main program.

Source code in expirebackups/expire.py
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
def main(argv=None):  # IGNORE:C0111
    """main program."""

    if argv is None:
        argv = sys.argv

    program_name = os.path.basename(sys.argv[0])
    program_version = "v%s" % __version__
    program_build_date = str(__updated__)
    program_version_message = "%%(prog)s %s (%s)" % (program_version, program_build_date)
    program_shortdesc = Version.description
    user_name = "Wolfgang Fahl"
    program_license = """%s

  Created by %s on %s.
  Copyright 2008-2022 Wolfgang Fahl. All rights reserved.

  Licensed under the Apache License 2.0
  http://www.apache.org/licenses/LICENSE-2.0

  Distributed on an "AS IS" basis without warranties
  or conditions of any kind, either express or implied.

USAGE
""" % (
        program_shortdesc,
        user_name,
        str(__date__),
    )

    try:
        # Setup argument parser
        parser = ArgumentParser(description=program_license, formatter_class=RawDescriptionHelpFormatter)
        parser.add_argument("-d", "--debug", dest="debug", action="store_true", help="show debug info")

        # expiration schedule selection
        parser.add_argument(
            "--days",
            type=int,
            default=defaultDays,
            help="number of consecutive days to keep a daily backup (default: %(default)s)",
        )
        parser.add_argument(
            "--weeks",
            type=int,
            default=defaultWeeks,
            help="number of consecutive weeks to keep a weekly backup (default: %(default)s)",
        )
        parser.add_argument(
            "--months",
            type=int,
            default=defaultMonths,
            help="number of consecutive month to keep a monthly backup (default: %(default)s)",
        )
        parser.add_argument(
            "--years",
            type=int,
            default=defaultYears,
            help="number of consecutive years to keep a yearly backup (default: %(default)s)",
        )

        # file filter selection arguments
        parser.add_argument(
            "--minFileSize",
            type=int,
            default=defaultMinFileSize,
            help="minimum File size in bytes to filter for (default: %(default)s)",
        )
        parser.add_argument("--rootPath", default=".")
        parser.add_argument("--baseName", default=None, help="the basename to filter for (default: %(default)s)")
        parser.add_argument("--ext", default=None, help="the extension to filter for (default: %(default)s)")

        parser.add_argument(
            "--createTestFiles",
            type=int,
            default=None,
            help="create the given number of temporary test files (default: %(default)s)",
        )

        parser.add_argument("-f", "--force", action="store_true")
        parser.add_argument("-V", "--version", action="version", version=program_version_message)

        args = parser.parse_args(argv[1:])
        if args.createTestFiles:
            path, _backupFiles = ExpireBackups.createTestFiles(args.createTestFiles)
            print(f"created {args.createTestFiles} test files with extension '.tst' in {path}")
            print(
                f"Please try out \nexpireBackups --rootPath {path} --baseName expireBackup --ext .tst --minFileSize 0"
            )
            print(
                "then try appending the -f option to the command that will actually delete files (which are in a temporary directory"
            )
            print(
                "and run the command another time with that option to see that no files are deleted any more on second run"
            )
        else:
            dryRun = True
            if args.force:
                dryRun = False
            expiration = Expiration(
                days=args.days,
                months=args.months,
                weeks=args.weeks,
                years=args.years,
                minFileSize=args.minFileSize,
                debug=args.debug,
            )
            eb = ExpireBackups(
                rootPath=args.rootPath,
                baseName=args.baseName,
                ext=args.ext,
                expiration=expiration,
                dryRun=dryRun,
                debug=args.debug,
            )
            eb.doexpire(args.force)

    except KeyboardInterrupt:
        ### handle keyboard interrupt ###
        return 1
    except Exception as e:
        if DEBUG:
            raise (e)
        indent = len(program_name) * " "
        sys.stderr.write(program_name + ": " + repr(e) + "\n")
        sys.stderr.write(indent + "  for help use --help")
        if args.debug:
            print(traceback.format_exc())
        return 2

version

Created on 2022-04-01

@author: wf

Version

Bases: object

Version handling for pyExpireBackups

Source code in expirebackups/version.py
 8
 9
10
11
12
13
14
15
16
17
class Version(object):
    """
    Version handling for pyExpireBackups
    """

    name = "pyExpireBackups"
    description = "Backup expiration based on rules (yearly,monthly,weekly,daily ...)"
    version = "0.1.0"
    date = "2022-04-01"
    updated = "2025-10-16"