luigi.contrib.external_daily_snapshot

Classes

ExternalDailySnapshot(*args, **kwargs)

Abstract class containing a helper method to fetch the latest snapshot.

class luigi.contrib.external_daily_snapshot.ExternalDailySnapshot(*args, **kwargs)[source]

Abstract class containing a helper method to fetch the latest snapshot.

Example:

class MyTask(luigi.Task):
  def requires(self):
    return PlaylistContent.latest()

All tasks subclassing ExternalDailySnapshot must have a luigi.DateParameter named date.

You can also provide additional parameters to the class and also configure lookback size.

Example:

ServiceLogs.latest(service="radio", lookback=21)

date

Parameter whose value is a date.

A DateParameter is a Date string formatted YYYY-MM-DD. For example, 2013-07-10 specifies July 10, 2013.

DateParameters are 90% of the time used to be interpolated into file system paths or the like. Here is a gentle reminder of how to interpolate date parameters into strings:

class MyTask(luigi.Task):
    date = luigi.DateParameter()

    def run(self):
        templated_path = "/my/path/to/my/dataset/{date:%Y/%m/%d}/"
        instantiated_path = templated_path.format(date=self.date)
        # print(instantiated_path) --> /my/path/to/my/dataset/2016/06/09/
        # ... use instantiated_path ...

To set this parameter to default to the current day. You can write code like this:

import datetime

class MyTask(luigi.Task):
    date = luigi.DateParameter(default=datetime.date.today())

classmethod latest(*args, **kwargs)[source]: This is cached so that requires() is deterministic.