Skip to content

Instantly share code, notes, and snippets.

@danizen
Last active February 7, 2018 21:55
Show Gist options
  • Save danizen/69685d652ca0ee1a25ac6593a3398e57 to your computer and use it in GitHub Desktop.
Save danizen/69685d652ca0ee1a25ac6593a3398e57 to your computer and use it in GitHub Desktop.
A target that causes luigi to run again after a file reaches a maximum age
class RecentlyModifiedLocalTarget(luigi.LocalTarget):
"""
The local file must have been recently enough modified to "exist", driving luigi's DAG
"""
def __init__(self, *args, **kwargs):
"""
Extract an optional maxage parameter
"""
maxage = kwargs.pop('maxage', None)
super(RecentlyModifiedLocalTarget, self).__init__(*args, **kwargs)
# Process the maximum age for units of days, weeks, and months
if maxage is None:
maxage = 5*24*60*60
if isinstance(maxage, str):
if maxage.endswith('d'):
maxage = int(maxage[:-1]) * 24*60*60
elif maxage.endswith('w'):
maxage = int(maxage[:-1]) * 7*24*60*60
elif maxage.endswith('m'):
maxage = int(maxage[:-1]) * 30*24*60*60
else:
maxage = int(maxage)
self.maxage = maxage
def exists(self):
"""
If the file exists, stat it
If its age relative to now is more that the maxage, then it doesn't exist
:return: True or False
"""
exists = super(RecentlyModifiedLocalTarget, self).exists()
if exists:
finfo = os.stat(self.path)
if time.time() - finfo.st_mtime > self.maxage:
return False
return exists
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment