kvnkho

import pandas as pd
from sklearn.preprocessing import minmax_scale

# %% tags=["parameters"]
# declare a list tasks whose products you want to use as inputs
upstream = ["extract"]
product = None
engine = None

# %% 

kvnkho

import pandas as pd

# %% tags=["parameters"]
# declare a list tasks whose products you want to use as inputs
upstream = None
product = None

# %%
df = pd.DataFrame({"col1": [1,2,3], "col2":[2,3,4]})
df.to_parquet(product["data"])

kvnkho

def loop(start_date, end_date):
    date_range = get_date_range(start_date, end_date)
    for date in date_range:
        e = extract(date)
        t = transform(e)
        l = load(t)

kvnkho

@flow
def loop_flow(start_date, end_date):
    date_range = get_date_range(start_date, end_date)
    for date in date_range:
        e = extract(date)
        t = transform(e)
        l = load(t)

kvnkho

# Prefect 2.0
@flow 
def get_item():
    e = extract().result()          # returns [1,2,3]
    e.pop(0)
    t = transform(e[1])
    l = load(t)

kvnkho

# Prefect 1.0
@task
def pop(x):
    x.pop(0)
    return x
  
with Flow("get-item") as flow:
    e = extract()        # returns [1,2,3]
    e2 = pop(e)
    t = transform(e2[1])

kvnkho

# Prefect 1.0 - Incorrect Usage
with Flow("get-item") as flow:
    e = extract()        # returns [1,2,3]
    e.pop(0)
    t = transform(e[1])
    l = load(t)

kvnkho

# Prefect 1.0
with Flow("get-item") as flow:
    e = extract()        # returns [1,2,3]
    t = transform(e[1])
    l = load()

kvnkho

# Prefect 2.0
@flow
def myflow():
    e = extract()
    cond = check_condition(e).result()
    if cond == True:
        t = transform1(e)
    else:
        t = transform2(e)
    l = load(t)

kvnkho

# Prefect 1.0 - Incorrect Usage
with Flow("basic") as flow:
    e = extract()
    cond = check_condition()
    t = transform1(e)
    with case(cond, True):
        t = transform2(t)
    l = load(t)