Justin Martin

03/07/2022, 2:34 PM
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
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.

Kevin Kho

03/07/2022, 2:37 PM
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:
Copy code
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

Justin Martin

03/07/2022, 2:41 PM
Copy 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')

def truncate_table(table_name) -> None:
	query=f'truncate table {table_name}’

def run_load_proc(prod) -> None:
	query=f'{exec {proc}’

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)

if __name__ == "__main__":

Kevin Kho

03/07/2022, 2:42 PM
Oh man I tagged the wrong guy hahaha. I edited it
Thanks for moving the traceback!
👍 1