ps import step

@step

def my_step() -> None:

...@pipeline

def my_pipeline(my_step):

my_step()

# Old: Create an instance of the pipeline and then call `pipeline_instance.configure(...)`

pipeline_instance = my_pipeline(my_step=my_step())

pipeline_instance.configure(enable_cache=False)

from zenml import pipeline, step

@step

def my_step() -> None:

...

@pipeline

def my_pipeline():

my_step()

# New: Call the `with_options(...)` method on the pipeline

my_pipeline = my_pipeline.with_options(enable_cache=False)

Running pipelines

from zenml.pipelines import pipeline

from zenml.steps import step

@step

def my_step() -> None:

...

@pipeline

def my_pipeline(my_step):

my_step()

# Old: Create an instance of the pipeline and then call `pipeline_instance.run(...)`

pipeline_instance = my_pipeline(my_step=my_step())

pipeline_instance.run(...)

from zenml import pipeline, step

@step

def my_step() -> None:

...

@pipeline

def my_pipeline():

my_step()

my_pipeline()  # New: Call the pipeline

Scheduling pipelines

from zenml.pipelines import pipeline, Schedule

from zenml.steps import step

@step

def my_step() -> None:

...

@pipeline

def my_pipeline(my_step):

my_step()

# Old: Create an instance of the pipeline and then call `pipeline_instance.run(schedule=...)`

schedule = Schedule(...)

pipeline_instance = my_pipeline(my_step=my_step())

pipeline_instance.run(schedule=schedule)

from zenml.pipelines import Schedule

from zenml import pipeline, step

@step

def my_step() -> None:

...

@pipeline

def my_pipeline():

my_step()

# New: Set the schedule using the `pipeline.with_options(...)` method and then run it

schedule = Schedule(...)

my_pipeline = my_pipeline.with_options(schedule=schedule)

my_pipeline()

Check out this page for more information on how to schedule your pipelines.

Fetching pipelines after execution

pipeline: PipelineView = zenml.post_execution.get_pipeline("first_pipeline")

last_run: PipelineRunView = pipeline.runs[0]

# OR: last_run = my_pipeline.get_runs()[0]