import prefect

import os

import pyodbc

from prefect import task, Flow

@task

def table_creation():

   

logger = prefect.context.get("logger")

   

conn = pyodbc.connect('Driver={SQL Server};' 'Server=localhost\sqlserver;' 'Database=test;' 'user=prova;' 'password=provaprova;')

   

cursor = conn.cursor()

   

file = open(r"C:\Users\andrea.nerla\Desktop\sidal\sql sidal 2\quote details table creazione.sql", 'r')

   

sql = file.read()

   

file.close()

   

cursor.execute(sql)

   

conn.commit()

   

cursor.close()

   

conn.close()

@task

 

def extract_load():

   

return os.system(r'"python C:\Users\andrea.nerla\Desktop\sidal\python_per_test\test_quotedetails.py"')

@task

 

def transform():

#    logger = prefect.context.get("logger")

#    conn = pyodbc.connect('Driver={SQL Server};' 'Server=localhost\sqlserver;' 'Database=test;' 'user=prova;' 'password=provaprova;')

#    cursor = conn.cursor()

   

file = open(r"C:\Users\andrea.nerla\Desktop\sidal\sql sidal 2\quote details stored proc.sql", 'r')

   

sql = file.read()

   

file.close()

#    cursor.execute(sql)

#    conn.commit()

#    conn.close()

with Flow("quotedetails_flow") as flow:

   

table_creation()

   

extract_load()

   

transform()

flow.register(project_name="quotedetails_test")

flow.run()

In your code, there are no dependencies between the 3 tasks, hence Prefect runs them in parallel or in random order. The piece of code I shared, show how to explicitly define dependecies between tasks using the imperative style.

More info here https://docs.prefect.io/core/concepts/flows.html