Implementation:Spotify Luigi ExternalDailySnapshot

Knowledge Sources	Spotify_Luigi
Domains	Data_Sources, Scheduling
Last Updated	2026-02-10 08:00 GMT

Overview

ExternalDailySnapshot is a Luigi contrib class that represents an external data source produced as daily snapshots. It provides a latest() class method that automatically finds the most recent available snapshot by looking back over a configurable number of days, making it easy to depend on external daily data without hardcoding dates.

Description

The ExternalDailySnapshot class extends luigi.ExternalTask and adds:

• A required date parameter (DateParameter) representing the snapshot date.
• A latest(cls, *args, **kwargs) class method that searches backward from a given date (default: today) over a lookback window (default: 14 days) to find the most recent snapshot where complete() returns True.

The latest() method is cached using an internal class-level cache (__cache) so that repeated calls within the same process (e.g., from multiple tasks calling requires()) return deterministic results without re-checking completeness.

If no completed snapshot is found within the lookback window, the method logs a debug message and returns the task instance for the oldest date checked (allowing Luigi to register it as an incomplete dependency).

Usage

Subclass ExternalDailySnapshot, implement output() to define the target for each date, and use MySnapshot.latest() in other tasks' requires() methods.

Code Reference

Source Location

luigi/contrib/external_daily_snapshot.py (75 lines)

Signature

class ExternalDailySnapshot(luigi.ExternalTask):
    date = luigi.DateParameter()

    @classmethod
    def latest(cls, *args, **kwargs):
        """
        Returns the most recent completed snapshot.

        Keyword arguments:
            date (datetime.date): Starting date to look back from (default: today).
            lookback (int): Number of days to look back (default: 14).
            *args, **kwargs: Additional arguments forwarded to the class constructor.

        Results are cached for deterministic requires() calls.
        """

Import

from luigi.contrib.external_daily_snapshot import ExternalDailySnapshot

I/O Contract

Inputs

Input	Type	Description
date	DateParameter	The date of the snapshot to check.
lookback	int (keyword to latest())	Number of days to search backward (default: 14).

Outputs

Output	Type	Description
self.output()	Target	Must be implemented by subclasses. Defines the target whose existence determines whether the snapshot for a given date is complete.
latest() return	ExternalDailySnapshot instance	The task instance for the most recent complete date, or the oldest checked date if none are complete.

Usage Examples

from luigi.contrib.external_daily_snapshot import ExternalDailySnapshot
import luigi

class PlaylistContent(ExternalDailySnapshot):
    def output(self):
        return luigi.contrib.hdfs.HdfsTarget(
            '/data/playlists/{}.csv'.format(self.date)
        )

class ServiceLogs(ExternalDailySnapshot):
    service = luigi.Parameter()

    def output(self):
        return luigi.contrib.gcs.GCSTarget(
            'gs://logs/{}/{}.jsonl'.format(self.service, self.date)
        )

class AnalyticsJob(luigi.Task):
    def requires(self):
        return {
            'playlists': PlaylistContent.latest(),
            'logs': ServiceLogs.latest(service='radio', lookback=21),
        }

    def run(self):
        playlists = self.input()['playlists']
        logs = self.input()['logs']
        # Process the most recent available snapshots...

Related Pages

• Spotify_Luigi_External_Data_Sources -- Principle governing external data source abstractions
• luigi.ExternalTask -- Parent class; tasks whose outputs are produced by an external process