Run part of a Flask python code once a day

Question

I have finally managed to create a postgres table by downloading info using ftp.

This db stors more then 11,000 stock symbols & other data, which take time to construct every time the function is called.

Is it possible that the function could run every morning at a specific time only, that the db will be updated?

Also, what will be the best location to write this function within the Flask app structure?

The function is:

def symbol_search():
    flo = BytesIO()

    directory = 'symboldirectory'
    filenames = ('otherlisted.txt', 'nasdaqlisted.txt')

    ftp = FTP('ftp.nasdaqtrader.com')
    ftp.login()
    ftp.cwd(directory)

    #Create pandas dataframes of stock symbols from the nasdaqlisted and otherlisted files.
    for item in filenames:
        nasdaq_exchange_info=[]
        ftp.retrbinary('RETR ' + item, flo.write)
        flo.seek(0)
        nasdaq_exchange_info.append(pd.read_fwf(flo))
    ftp.quit()

    nasdaq_exchange_info=pd.concat(nasdaq_exchange_info, axis=1)
    nasdaq_exchange_info[['symbol', 'name', 'Exchange', 'Symbol', 'etf', 'Lot_size', 'Test', 'NASDAQ_Symbol']]=nasdaq_exchange_info['ACT Symbol|Security Name|Exchange|CQS Symbol|ETF|Round Lot Size|Test Issue|NASDAQ Symbol'].str.split('|', expand=True)
    nasdaq_exchange_info=nasdaq_exchange_info.drop(nasdaq_exchange_info.columns[[0]], axis=1).dropna()
    nasdaq_exchange_info=nasdaq_exchange_info[nasdaq_exchange_info.Test != 'Y']
    nasdaq_exchange_info=nasdaq_exchange_info[nasdaq_exchange_info.symbol != 'Y']
    nasdaq_exchange_info=nasdaq_exchange_info[~nasdaq_exchange_info.symbol.str.contains('symbol')]
    nasdaq_exchange_info=nasdaq_exchange_info[~nasdaq_exchange_info.symbol.str.contains('File')]
    nasdaq_exchange_info=nasdaq_exchange_info[~nasdaq_exchange_info.name.str.contains('%')]
    nasdaq_exchange_info=nasdaq_exchange_info[~nasdaq_exchange_info.name.str.contains('arrant')]
    nasdaq_exchange_info=nasdaq_exchange_info.drop(['Symbol', 'Exchange', 'Lot_size', 'Test', 'NASDAQ_Symbol', 'etf'], axis = 1)
    nasdaq_exchange_info=nasdaq_exchange_info[['name', 'symbol']].values.tolist()
    return nasdaq_exchange_info

Answer 1

You can use the APScheduler library.

from apscheduler.schedulers.blocking import BlockingScheduler
#...
scheduler = BackgroundScheduler(timezone="Europe/Rome")
# Runs from Monday to Friday at 5:30 (am)
scheduler.add_job(
    func=search,
    trigger="cron",
    max_instances=1,
    day_of_week='mon-fri',
    hour=5,
    minute=30
)
scheduler.start()

Take a look at this link to better understand how to set the triggers to call the function