Hello all, thanks for building such an amazing product! I do have a situation that i could use some help with: i have a task that does something very small; is there way to define a FunctionTask on the fly and add it to the flow without having to build an actual function with the

@task

decorator? Here is an extremely simplified version of what i'm doing (See thread). Am I able to define the run_load_proc function as just a FUnctionTask within the flow without an explicit function? Also, not sure if defining the global SqlServerExecute is a total anti-pattern, let me know. Thanks for all of your help.

Hi @Justin Martin, could you move the traceback to the thread when you get a chance to keep the main channel a bit neater? You can use a lambda. The global SQLServerExecute is fine. The lambda looks something like:

with Flow(..) as flow:
     a = get_data() # task that returns 1
     task(lambda x: x+1)(a)

I think this will work, but you don’t get task names so I personally don’t like it

Code:

import prefect
from prefect import task, Flow
from sqlserver import SqlServerExecute

exec_sql = SqlServerExecute(db_name='FinanceDM', host='fstsqldev', driver='{SQL Server Native Client 11.0}; Trusted_Connection=yes;' commit=True, user='jumartin')


@task
def truncate_table(table_name) -> None:
	query=f'truncate table {table_name}’
	exec_sql.run(query=query


@task
def run_load_proc(prod) -> None:
	query=f'{exec {proc}’
 	exec_sql.run(query=query)


def main():
	with Flow("Load Data") as flow:
		delete_table = truncate_table('my_table_name')
		run_proc = run_load_proc(proc='dbo.my_proc_name', upstream_tasks=delete_table)
	
	flow.run()

if __name__ == "__main__":
	main()

Oh man I tagged the wrong guy hahaha. I edited it