Example of historical stock price update (Rocketry)

// main.py

from rocketry import Rocketry
from rocketry.conds import cron
from rocketry.conds import after_success, after_fail, after_finish
from rocketry.conds import after_all_success, after_any_success, after_any_finish, after_any_fail
from rocketry.args import Return
import yfinance as yf
import db
import email

app = Rocketry()

@app.task(cron(minute="0", hour="6", day_of_week="1-5"))
def get_symbols() -> list:
    symbols = db.get_symbols()
    return symbols

@app.task(after_success(get_symbols))
def get_price_history(symbols = Return(get_symbols)):
    price_history_list = []
    
    for symbol in symbols:
        data = yf.Ticker[symbol].history(period="max")
        price_history_list.append(data)
        
    return price_history_list

@app.task(after_success(get_price_history))
def insert_db(data = Return(get_price_history)):
    for stock in data:
        for row in stock.to_dict('records'):
            db.insert(row)
    
@app.task(after_all_success(get_symbols, get_price_history, insert_db))
def send_success_email():
    email.send(title = "Historical price update successfully")

@app.task(after_any_fail(get_symbols, get_price_history, insert_db))
def send_fail_email():
    email.send(title = "Historical price update failed")


if __name__ == '__main__':
    app.run()