Miksus / rocketry

Modern scheduling library for Python
https://rocketry.readthedocs.io
MIT License
3.27k stars 108 forks source link

ENH: Create task from class directly #3

Closed Miksus closed 2 years ago

Miksus commented 3 years ago

Is your feature request related to a problem? Please describe. Proposal for new task class: create a task directly from a class. Requires possibly a new metaclass. The task will be created when the class itself is initiated.

from redengine.tasks import ClassTask

class MyTask(ClassTask):
    name = 'my-task'
    start_cond = 'daily'
    execution = 'process'

    def __init__(self, session):
        ... # Do whatever

    def execute(self):
        ... # Do whatever

Why? You can bundle all the functions related to the task nicely to one place.

Example, now one can do:

from redengine.tasks import ClassTask

def extract():
    ...

def transform(data):
    ...
    return data

def load(data):
    ...

@FuncTask(name="etl", start_cond="daily")
def process_etl():
    data = extract()
    data = transform(data)
    load(data)

Proposed:

from redengine.tasks import ClassTask

class ProcessETL(ClassTask):
    name = 'etl'
    start_cond = 'daily'

    def execute(self):
        data = self.extract()
        data = self.transform(data)
        self.load(data)

    def extract(self):
        ....

    def transform(self, data):
        ....
        return data

    def load(self, data):
        ...