HOw to use LookUP Tables to Simplify Flask Routes

lookup_jobs.py

import asyncio
from typing import Callable, Coroutine, Dict

from flask import Blueprint, jsonify

from src.authentication.cron_validator import cron_validator
from src.exceptions import InputError, RequestError, status_codes
from src.external_data.StocklistAPI.update_stock_lists import UpdateStocks
from src.external_data.eod.analystrankingsfundamental import \
    AnalystRankingsDataSource
from src.external_data.eod.datasourcedividendsbyyear import \
    DataSourceDividendsByYear
from src.external_data.eod.earningsfundamental import EarningsDataSource
from src.external_data.eod.financialsfundamentals import FinancialsDataSource
from src.external_data.eod.generalfundametal import DataSourceGeneral
from src.external_data.eod.highlightsfundamental import DataSourceHighlights
from src.external_data.eod.holdersfundamentals import HoldersDataSource
from src.external_data.eod.insidertransactionsfundamentals import \
    InsiderTradingDataSource
from src.external_data.eod.outstandingsharesfundamental import \
    OutStandingSharesDataSource
from src.external_data.eod.sharestatsfundamental import DataSourceShareStats
from src.external_data.eod.splitdividendsfundamental import \
    DataSourceSplitDividends
from src.external_data.eod.valuationsfundamental import DataSourceValuations
from src.external_data.eod_stock.eod_stock_data import BulkEndOfDayStockSource
from src.external_data.prepared_data.github_stock_data import GithubStockData
from src.external_data.updates.update_fundamentals import UpdateFundamentals
from src.external_data.yahoo.eod_yahoo import YahooEODData
from src.external_data.yahoo.news_yahoo_finance import YahooFinanceNews, CuratedStocksNewsDownloader
from src.external_data.yahoo.yahoo_fundamental import CreateFundamentalYahoo
from src.main.loader import eod_data_source, logger
from src.social_media.twitter.news import CronNewsTwitter

jobs_cron_route = Blueprint('cron_jobs', __name__)

# NOTE data sources are Singletons
CACHE_TIMEOUT_HOUR = 60 * 60
CACHE_TIMEOUT_TWELVE = 60 * 60 * 12
CACHE_TIMEOUT_DAY = 60 * 60 * 24
CACHE_TIMEOUT_TEN_MINUTES = 10 * 60

CompanyType = dict[str, str | int | dict[str, list[dict[str, int | str]]]]


# Note the cache will return the function not the function results
def _cron_jobs_routes() -> Dict[str, Callable]:
    """Look Up Table for _cron routes"""
    data_source: DataSourceGeneral = DataSourceGeneral()
    github_stock_data = GithubStockData()
    yahoo_eod_data = YahooEODData()
    yahoo_news_data = YahooFinanceNews()
    yahoo_fundamentals = CreateFundamentalYahoo()
    updates = UpdateFundamentals()
    tweets = CronNewsTwitter()
    news_curation = CuratedStocksNewsDownloader()
    stocks_api = UpdateStocks()

    return {'exchanges': data_source.update_exchanges_from_eod,
            'tickers': data_source.update_tickers,
            'stocks': stocks_api.update,
            'parse-github': github_stock_data.start_parsing,
            'parse-eod-historical-yahoo': yahoo_eod_data.create_eod_historical_data,
            'parse-eod-daily-yahoo': yahoo_eod_data.create_eod_daily_data,
            'parse-news-yahoo': yahoo_news_data.parse_news_yahoo,
            'fundamentals-yahoo': yahoo_fundamentals.create_fundamental_from_yahoo,
            'update-fundamentals': updates.run_updates,
            'tweet-news': tweets.create_tweets,
            'tweet-sentiment': tweets.tweet_sentiments_stocks,
            'compile-finance-tweeter': tweets.compile_finance_tweeter,
            'tldr-news': tweets.run_tldr_on_articles_and_update,
            'prep-articles': tweets.prep_articles_for_sentiment_analysis,
            'download-articles': tweets.download_articles_and_update,
            'curated-news': news_curation.create_curation_news}


class InvalidPath(Exception):
    """error will be thrown when cron tries to execute an invalidPath"""

    def __init__(self, description: str, path: str):
        self.description = description
        self.path = path


def _fundamental_routes() -> Dict[str, Callable[[], Coroutine]]:
    """Look Up Table for _cron routes
    main jobs for fundamentals 
    """

    general_fundamentals: DataSourceGeneral = DataSourceGeneral()
    highlights_source: DataSourceHighlights = DataSourceHighlights()
    valuations_source: DataSourceValuations = DataSourceValuations()
    share_stats_source: DataSourceShareStats = DataSourceShareStats()
    split_dividends_source: DataSourceSplitDividends = DataSourceSplitDividends()
    dividends_year_source: DataSourceDividendsByYear = DataSourceDividendsByYear()
    analyst_ranking_source: AnalystRankingsDataSource = AnalystRankingsDataSource()
    holders_source: HoldersDataSource = HoldersDataSource()
    insider_source: InsiderTradingDataSource = InsiderTradingDataSource()
    outstanding_source: OutStandingSharesDataSource = OutStandingSharesDataSource()
    earnings_source: EarningsDataSource = EarningsDataSource()
    financials_source: FinancialsDataSource = FinancialsDataSource()
    daily_eod_data: BulkEndOfDayStockSource = BulkEndOfDayStockSource()

    return {'get': eod_data_source.set_fundamental_data,
            'set-general': general_fundamentals.save_general_data_to_datastore,
            'set-highlights': highlights_source.set_fundamental_highlights,
            'set-valuations': valuations_source.set_valuation_data,
            'set-share-stats': share_stats_source.set_share_stats,
            'set-split-dividends': split_dividends_source.set_split_dividends,
            'set-dividends': dividends_year_source.set_dividends_by_year,
            'set-analyst-rankings': analyst_ranking_source.set_analyst_rankings,
            'set-funds-holders': holders_source.set_funds_holders,
            'set-institution-holders': holders_source.set_institution_holders,
            'set-insider-trading': insider_source.set_insider_trading,
            'set-annual-outstanding-shares': outstanding_source.set_annual_out_standing_shares,
            'set-quarterly-outstanding-shares': outstanding_source.set_quarterly_out_standing,
            'set-history-earnings': earnings_source.set_history_earnings,
            'set-trend-earnings': earnings_source.set_trend_earnings,
            'set-annual-earnings': earnings_source.set_annual_earnings,
            'set-quarterly-balance-sheet': financials_source.set_quarterly_balance_sheet,
            'set-annual-balance-sheet': financials_source.set_yearly_balance_sheets,
            'daily-eod-data': daily_eod_data.bulk_update_eod_tickers}


@jobs_cron_route.route("/_cron/jobs/<string:path>", methods=['GET'])
@cron_validator
def cron_jobs(path: str) -> tuple:
    """cron jobs for executing external API requests"""
    if not path:
        raise InvalidPath(description="please supply a valid path", path=path)
    try:
        logger.warning('cron jobs path: %s', path)
        _function: Callable = _cron_jobs_routes()[path]
    except KeyError as e:
        raise InvalidPath(description="please supply a valid path", path=path) from e

    # Runs the retrieved function
    response_data = asyncio.run(_function())

    return jsonify(dict(status=True, payload=response_data)), status_codes.status_ok_code


@jobs_cron_route.route("/_cron/jobs/fundamentals/<string:path>", methods=['GET'])
@cron_validator
def fundamentals_jobs(path: str) -> tuple:
    """
    could merge this two methods / routes
    cron jobs for executing external API requests"""
    if not path:
        raise RequestError(description="path is invalid", url=path)
    try:
        logger.warning('cron jobs path: %s', path)
        asyncio.run(_fundamental_routes()[path]())
    except KeyError as e:
        raise InputError(description="path is not available") from e

    return dict(status=True, message='successfully updated fundamental data'), status_codes.status_ok_code

# TODO create a CLI for EOD where i can pass a date range