@prefect.task
def fetch_invalid_reward_transactions(batch_size, last_pid=None):
    logger = get_logger()
    
    data = list(dbconn.coll_name.find(
        filter=query
    ).sort("_id", 1).limit(int(batch_size)))

    return data

@prefect.task
def filter_unique_user_txns(txn):
    result = []
    for txn in txn:
        # do something
    
        result.append({})
    return result
    

@prefect.flow(name="test flow", task_runner=SequentialTaskRunner())
def flow(batch_size: int = 500):
    last_pid = fetch_last_id()

    invalid_transactions = fetch_invalid_reward_transactions(
        batch_size,
        last_pid=last_pid
    )

    formatted_invalid_transactions = filter_unique_user_txns(
        txn=invalid_transactions
    )

    save_result = save_to_db(
        formatted_invalid_transactions
    )