feat: Add logic to be able to create pipelines on naas.

Dr0p42 commented 1 year ago

This pull requests resolves: https://github.com/jupyter-naas/naas/issues/344

It's adding logic to easily create pipeline on naas.

It's that easy:

from naas.pipelines.pipelines import Pipeline, DummyStep, DummyErrorStep, NotebookStep, End, ParallelStep

pipeline = Pipeline()

success_notebook = NotebookStep('Success notebook', 'success_notebook.ipynb')

failing_notebook = NotebookStep('Failing notebook', 'failing_notebook.ipynb')

pipeline >> failing_notebook >> success_notebook >> End()

failing_notebook.on_error >> DummyStep('My Alert') >> DummyStep('Cleaning step')

pipeline.run()

This will give you the following

And after running it:

We can also run it by using pipeline.run('progress') which will show something like this:

In this example we have a notebook failing on purpose to trigger the on_error step.

Some other examples:

Imagine you have a bunch of notebooks to download videos from Youtube and compute a transcript out of it:

from naas.pipelines.pipelines import Pipeline, DummyStep, DummyErrorStep, NotebookStep, End, ParallelStep

pipeline = Pipeline()

download = DummyStep('Download Youtube Video')

download.on_error >> DummyStep('Alert youtube download error')

pipeline >> download >> DummyStep('Convert video to mp3') \
    >> DummyStep('Upload mp3 to s3') \
    >> DummyStep('Start Transcription') \
    >> DummyStep('Store transcript on S3') \
    >> [
        DummyStep('Notify Slack'),
        DummyStep('Notify Email')
    ] \
>> End()

pipeline

Here I am using DummyStep to give an idea, but it could definitely be NotebookSteps

Now imagine you want to fetch a bunch of ticker/stocks and do some computation with it and ultimately buy or sell. You could fetch the ticker at the same time and also extract sentiment from twitter about the companies.

from naas.pipelines.pipelines import Pipeline, DummyStep, DummyErrorStep, NotebookStep, End, ParallelStep

pipeline = Pipeline()

auto_buy_sell = DummyStep('Auto buy/sell stocks')

pipeline >> [
        DummyStep('Get Tesla Stock'),
        DummyStep('Get Google Stock'),
        DummyStep('Get Amazon Stock'),
        DummyStep('Get Microsoft Stock'),
        DummyStep('Get Meta Stock'),

        DummyStep('Get Tesla Sentiment from Twitter'),
        DummyStep('Get Google Sentiment from Twitter'),
        DummyStep('Get Amazon Sentiment from Twitter'),
        DummyStep('Get Microsoft Sentiment from Twitter'),
        DummyStep('Get Meta Sentiment from Twitter'),
    ] \
    >> DummyStep('Compute overall performance') \
    >> auto_buy_sell\
    >> DummyStep('Notify completion') \
>> End()

# On error on this step we may want to alert someone to take a manual action quickly.
auto_buy_sell.on_error >> [
    DummyStep('Alert Slack'),
    DummyStep('Alert SMS'),
    DummyStep('Alert Whatsapp'),
    DummyStep('Alert PagerDuty'),
    DummyStep('Alert Email'),
]

pipeline