insert_customer = PostgresExecute(
    query=""" 
        INSERT INTO temp_deleted_customers (customer_id, synapse_account_id, date_deleted)
        VALUES (%s, %s, %s);
    """,
    db_name="xx",
    user="xx",
    host="xx",
    commit=True,
)


with Flow("Deleted Customer Nodes") as flow:
    deleted_customers = get_deleted_customers() # BigQueryTask
    closed_nodes = get_customer_deposit_nodes.map(deleted_customers) # Returns a (x,y,z) tuple per customer
    create_table = create_pg_table(
        password=PrefectSecret("POSTGRES_PASSWORD"), commit=True
    )
    insert_customer.map(
        data=closed_nodes, password=unmapped(PrefectSecret("POSTGRES_PASSWORD"))
    )

.to_dataframe

.to_sql

@task
def save_customers_to_sql(deleted_nodes, postgres_user, postgres_password):
    logger = prefect.context.get("logger")
    df = pd.DataFrame(deleted_nodes)
    df = df.dropna()
    engine = create_engine(
        f"postgresql+psycopg2://{postgres_user}:{postgres_password}@127.0.0.1/xxx",
        pool_recycle=3600,
    )
    conn = engine.connect()
    try:
        df.to_sql("temp_deleted_customers", conn, if_exists="replace")
    finally:
        <http://logger.info|logger.info>("Table has been created successfully")
        conn.close()

deleted_rows