upstream_task

from prefect import Flow, task, Task
import time
from prefect.executors import LocalDaskExecutor
from datetime import timedelta

@task()
def abc(x):
    return x

class B(Task):
    def run(self, x):
        return x

with Flow(
    "test",
) as flow:
    test = abc(1)
    test2 = B()(1, upstream_tasks=[test])

flow.run()

time.sleep(x)

@task()
def abc(x):
    pass

@task
def get_datestr():
  return prefect.context.today

create_tmp = Create_Tmp()
extract = Extract()
load = Load()
drop = Drop_tmp()

with Flow(..) as flow:
  datestr = get_datestr()
  
  shipcosts = create_tmp('ShippingFees_SQL\\Create_shipcostsTemp.sql', datestr)
  shiptotals = create_tmp('ShippingFees_SQL\\Create_shiptotalsTemp.sql', datestr)

  df = extract('ShippingFees_SQL\\GetShippingFeesByOrderItem.sql', datestr)
  df.set_upstream(shipcosts)
  df.set_upstream(shiptotals)

  load_fees = load(df, 'oFeesByOrderItem')

  run_drop = drop('ShippingFees_SQL\\Drop_shippingTemp.sql')
  run_drop.set_upstream(load_fees)

test2 = B()(1, upstream_tasks=[test])

kwargs