import yaml from dataclasses import dataclass from typing import Dict import copy import logging @dataclass class ExportModulePair: """ Data structure pairing data export endpoints across adjacent fiscal periods. This class groups the URLs for the current and previous fiscal years, allowing the application to easily pass paired historical and active data sources to downstream report pipelines. """ current_fy:str | None = None prev_fy:str | None = None # Export module configuration file constants EXPORT_MODULE_URLS_KEY = "export_module_urls" EXPORT_MODULE_CLIENT_LIST_KEY = "clients_list" EXPORT_MODULE_NBS_KEY = "new_business_starts" EXPORT_MODULE_FUNDING_KEY = "capital_funding" EXPORT_MODULE_TRAININGS_KEY = "trainings" CURRENT_FY_KEY = "current_fy" PREV_FY_KEY = "prev_fy" LOGGING_CONFIG_SECTION_KEY = "logging" LOGGING_LEVEL_KEY = "level" LOGGING_FILE_KEY = "file" ERRORS_SECTION_KEY = "errors" CONTACT_NAME_KEY = "contact_name" CONTACT_METHOD_KEY = "contact_method" USDA_API_SECTION_KEY = "usda_api" USDA_API_KEY_KEY = "key" CENSUS_SECTION_KEY = "census" CENSUS_YEAR_KEY = "year" TEMPLATE_DICT = { EXPORT_MODULE_URLS_KEY: { EXPORT_MODULE_CLIENT_LIST_KEY: { CURRENT_FY_KEY: "", PREV_FY_KEY: "" }, EXPORT_MODULE_NBS_KEY: { CURRENT_FY_KEY: "", PREV_FY_KEY: "" }, EXPORT_MODULE_FUNDING_KEY: { CURRENT_FY_KEY: "", PREV_FY_KEY: "" }, EXPORT_MODULE_TRAININGS_KEY: { CURRENT_FY_KEY: "", PREV_FY_KEY: "" } }, LOGGING_CONFIG_SECTION_KEY: { LOGGING_LEVEL_KEY: "WARNING", LOGGING_FILE_KEY: "dashboard_log.log" }, ERRORS_SECTION_KEY: { CONTACT_NAME_KEY: "", CONTACT_METHOD_KEY: "", }, USDA_API_SECTION_KEY: { USDA_API_KEY_KEY: "", }, CENSUS_SECTION_KEY: { CENSUS_YEAR_KEY: "", } } class DashboardConfig: """ Central manager for application configuration state and schema validation. This class handles the serialization, parsing, and strict validation of the YAML configuration file. It acts as the single source of truth for all external Neoserra data endpoints and logging behaviors, abstracting the dictionary traversal away from the reporting classes. :param filename: The filepath to the target YAML configuration file. :type filename: str """ def __init__(self, filename:str): """ Initializes the configuration manager with a default structural template. Establishes the base dictionary structure in memory before attempting any file I/O operations, ensuring internal accessors do not fail on missing root keys. :param filename: The path to the configuration file on disk. :type filename: str """ self.config_dictionary:Dict = copy.deepcopy(TEMPLATE_DICT) self.filename = filename def set_clients_list_current_fy_url(self, current_fy_url:str): """ Updates the in-memory endpoint URL for the current fiscal period's client dataset. """ self.config_dictionary[EXPORT_MODULE_URLS_KEY][EXPORT_MODULE_CLIENT_LIST_KEY][CURRENT_FY_KEY] = current_fy_url def set_clients_list_prev_fy_url(self, prev_fy_url:str): """ Updates the in-memory endpoint URL for the previous fiscal period's client dataset. """ self.config_dictionary[EXPORT_MODULE_URLS_KEY][EXPORT_MODULE_CLIENT_LIST_KEY][PREV_FY_KEY] = prev_fy_url def get_clients_list_urls(self) -> ExportModulePair: """ Retrieves the paired endpoints for the client list dataset. :return: An object containing the current and previous fiscal period URLs. :rtype: ExportModulePair """ return ExportModulePair( current_fy=self.config_dictionary[EXPORT_MODULE_URLS_KEY][EXPORT_MODULE_CLIENT_LIST_KEY][CURRENT_FY_KEY], prev_fy=self.config_dictionary[EXPORT_MODULE_URLS_KEY][EXPORT_MODULE_CLIENT_LIST_KEY][PREV_FY_KEY] ) def set_nbs_milestones_current_fy_url(self, current_fy_url:str): """ Updates the in-memory endpoint URL for the current fiscal period's New Business Starts dataset. """ self.config_dictionary[EXPORT_MODULE_URLS_KEY][EXPORT_MODULE_NBS_KEY][CURRENT_FY_KEY] = current_fy_url def set_nbs_milestones_prev_fy_url(self, prev_fy_url:str): """ Updates the in-memory endpoint URL for the previous fiscal period's New Business Starts dataset. """ self.config_dictionary[EXPORT_MODULE_URLS_KEY][EXPORT_MODULE_NBS_KEY][PREV_FY_KEY] = prev_fy_url def get_nbs_milestones_urls(self) -> ExportModulePair: """ Retrieves the paired endpoints for the New Business Starts dataset. :return: An object containing the current and previous fiscal period URLs. :rtype: ExportModulePair """ return ExportModulePair( current_fy=self.config_dictionary[EXPORT_MODULE_URLS_KEY][EXPORT_MODULE_NBS_KEY][CURRENT_FY_KEY], prev_fy=self.config_dictionary[EXPORT_MODULE_URLS_KEY][EXPORT_MODULE_NBS_KEY][PREV_FY_KEY] ) def set_funding_milestones_current_fy_url(self, current_fy_url:str): """ Updates the in-memory endpoint URL for the current fiscal period's funding milestones dataset. """ self.config_dictionary[EXPORT_MODULE_URLS_KEY][EXPORT_MODULE_FUNDING_KEY][CURRENT_FY_KEY] = current_fy_url def set_funding_milestones_prev_fy_url(self, prev_fy_url:str): """ Updates the in-memory endpoint URL for the previous fiscal period's funding milestones dataset. """ self.config_dictionary[EXPORT_MODULE_URLS_KEY][EXPORT_MODULE_FUNDING_KEY][PREV_FY_KEY] = prev_fy_url def get_funding_milestones_urls(self) -> ExportModulePair: """ Retrieves the paired endpoints for the funding milestones dataset. :return: An object containing the current and previous fiscal period URLs. :rtype: ExportModulePair """ return ExportModulePair( current_fy=self.config_dictionary[EXPORT_MODULE_URLS_KEY][EXPORT_MODULE_FUNDING_KEY][CURRENT_FY_KEY], prev_fy=self.config_dictionary[EXPORT_MODULE_URLS_KEY][EXPORT_MODULE_FUNDING_KEY][PREV_FY_KEY] ) def set_trainings_current_fy_url(self, current_fy_url:str): """ Updates the in-memory endpoint URL for the current fiscal period's training events dataset. """ self.config_dictionary[EXPORT_MODULE_URLS_KEY][EXPORT_MODULE_TRAININGS_KEY][CURRENT_FY_KEY] = current_fy_url def set_trainings_prev_fy_url(self, prev_fy_url:str): """ Updates the in-memory endpoint URL for the previous fiscal period's training events dataset. """ self.config_dictionary[EXPORT_MODULE_URLS_KEY][EXPORT_MODULE_TRAININGS_KEY][PREV_FY_KEY] = prev_fy_url def get_trainings_urls(self) -> ExportModulePair: """ Retrieves the paired endpoints for the training events dataset. :return: An object containing the current and previous fiscal period URLs. :rtype: ExportModulePair """ return ExportModulePair( current_fy=self.config_dictionary[EXPORT_MODULE_URLS_KEY][EXPORT_MODULE_TRAININGS_KEY][CURRENT_FY_KEY], prev_fy=self.config_dictionary[EXPORT_MODULE_URLS_KEY][EXPORT_MODULE_TRAININGS_KEY][PREV_FY_KEY] ) def set_usda_api_key(self, key:str): self.config_dictionary[USDA_API_SECTION_KEY][USDA_API_KEY_KEY] = key def get_usda_api_key(self): return self.config_dictionary[USDA_API_SECTION_KEY][USDA_API_KEY_KEY] def set_census_year(self, year:str): self.config_dictionary[CENSUS_SECTION_KEY][CENSUS_YEAR_KEY] = year def get_census_year(self): return self.config_dictionary[CENSUS_SECTION_KEY][CENSUS_YEAR_KEY] def get_log_path(self) -> str | None: """ Retrieves the configured file path for the application's rotating file logger. :return: The target filepath, or None if the configuration is missing. :rtype: str | None """ try: return self.config_dictionary[LOGGING_CONFIG_SECTION_KEY][LOGGING_FILE_KEY] except KeyError: return None def get_log_level(self) -> int: """ Translates the human-readable logging level from the configuration file into system constants. Maps string values (e.g., "DEBUG", "WARNING") to their corresponding integer values required by the native Python `logging` module. :return: The integer constant representing the logging severity level. :rtype: int """ try: log_level = self.config_dictionary[LOGGING_CONFIG_SECTION_KEY][LOGGING_LEVEL_KEY] except KeyError: return None if log_level == "DEBUG": return logging.DEBUG elif log_level == "INFO": return logging.INFO elif log_level == "WARNING" or log_level == "WARN": return logging.WARNING elif log_level == "ERROR": return logging.ERROR elif log_level == "CRITICAL": return logging.CRITICAL else: return None def set_errors_contact_name(self, name:str): self.config_dictionary[ERRORS_SECTION_KEY][CONTACT_NAME_KEY] = name def get_errors_contact_name(self) -> str: return self.config_dictionary[ERRORS_SECTION_KEY][CONTACT_NAME_KEY] def set_errors_contact_method(self, method:str): self.config_dictionary[ERRORS_SECTION_KEY][CONTACT_METHOD_KEY] = method def get_errors_contact_method(self): return self.config_dictionary[ERRORS_SECTION_KEY][CONTACT_METHOD_KEY] def get_errors_contact_string(self) -> str: return f"Contact {self.get_errors_contact_name()} ({self.get_errors_contact_method()})" @staticmethod def write_template(filename:str): """ Bootstraps a new environment by generating a default configuration file on disk. Writes the hardcoded `TEMPLATE_DICT` schema to the specified path to ensure administrators have the correct structural format when deploying the dashboard for the first time. :param filename: The target filepath for the generated YAML file. :type filename: str """ with open(filename, "w") as stream: try: yaml.dump(TEMPLATE_DICT, stream, default_flow_style=False, indent=4) except Exception as e: raise IOError("Could not dump configuration template") from e def save(self): """ Serializes the current in-memory configuration state to the YAML file on disk. """ with open(self.filename, "w") as stream: try: yaml.dump(self.config_dictionary, stream, default_flow_style=False, indent=4) except Exception as e: raise IOError("Could not save configuration file!") from e def load(self): """ Reads the configuration file from disk and enforces strict schema validation. Acts as a fail-fast gateway for the application environment. It manually verifies the existence of every required organizational key (export modules, fiscal periods, and logging parameters). If the YAML file is structurally malformed or missing endpoints, it immediately raises explicit KeyErrors to halt application boot, preventing cascading fetch failures later. """ try: with open(self.filename, "r") as stream: config = yaml.safe_load(stream) self.config_dictionary = config export_module_urls = config.get(EXPORT_MODULE_URLS_KEY, None) # Parse all of the export module urls form the config file if export_module_urls is None: raise KeyError(f"The dashboard configuration file {self.filename} did not have config section {EXPORT_MODULE_URLS_KEY}.") # Handle the clients list section clients_section = export_module_urls.get(EXPORT_MODULE_CLIENT_LIST_KEY, None) if clients_section is None: raise KeyError(f"The dashboard configuration file {self.filename} did not have config section {EXPORT_MODULE_CLIENT_LIST_KEY} under section {EXPORT_MODULE_URLS_KEY}.") clients_section_current = clients_section.get(CURRENT_FY_KEY, None) clients_section_prev = clients_section.get(PREV_FY_KEY, None) if clients_section_current is None or clients_section_prev is None: raise KeyError(f"The dashboard configuration file {self.filename} did not have either a current fiscal year or a previous fiscal year section underneath section {EXPORT_MODULE_URLS_KEY}/{EXPORT_MODULE_CLIENT_LIST_KEY}. Add '{CURRENT_FY_KEY}:' and '{PREV_FY_KEY}:' under this config section.") nbs_section = export_module_urls.get(EXPORT_MODULE_NBS_KEY, None) if nbs_section is None: raise KeyError(f"The dashboard configuration file {self.filename} did not have config section {EXPORT_MODULE_NBS_KEY} under section {EXPORT_MODULE_URLS_KEY}.") nbs_section_current = nbs_section.get(CURRENT_FY_KEY, None) nbs_section_prev = nbs_section.get(PREV_FY_KEY, None) if nbs_section_current is None or nbs_section_prev is None: raise KeyError(f"The dashboard configuration file {self.filename} did not have either a current fiscal year or a previous fiscal year section underneath section {EXPORT_MODULE_URLS_KEY}/{EXPORT_MODULE_NBS_KEY}. Add '{CURRENT_FY_KEY}:' and '{PREV_FY_KEY}:' under this config section.") funding_section = export_module_urls.get(EXPORT_MODULE_FUNDING_KEY, None) if funding_section is None: raise KeyError(f"The dashboard configuration file {self.filename} did not have config section {EXPORT_MODULE_FUNDING_KEY} under section {EXPORT_MODULE_URLS_KEY}.") funding_section_current = funding_section.get(CURRENT_FY_KEY, None) funding_section_prev = funding_section.get(PREV_FY_KEY, None) if funding_section_current is None or funding_section_prev is None: raise KeyError(f"The dashboard configuration file {self.filename} did not have either a current fiscal year or a previous fiscal year section underneath section {EXPORT_MODULE_URLS_KEY}/{EXPORT_MODULE_FUNDING_KEY}. Add '{CURRENT_FY_KEY}:' and '{PREV_FY_KEY}:' under this config section.") trainings_section = export_module_urls.get(EXPORT_MODULE_TRAININGS_KEY, None) if trainings_section is None: raise KeyError(f"The dashboard configuration file {self.filename} did not have config section {EXPORT_MODULE_TRAININGS_KEY} under section {EXPORT_MODULE_URLS_KEY}.") trainings_section_current = trainings_section.get(CURRENT_FY_KEY, None) trainings_section_prev = trainings_section.get(PREV_FY_KEY, None) if trainings_section_current is None or trainings_section_prev is None: raise KeyError(f"The dashboard configuration file {self.filename} did not have either a current fiscal year or a previous fiscal year section underneath section {EXPORT_MODULE_URLS_KEY}/{EXPORT_MODULE_TRAININGS_KEY}. Add '{CURRENT_FY_KEY}:' and '{PREV_FY_KEY}:' under this config section.") logging_section = self.config_dictionary.get(LOGGING_CONFIG_SECTION_KEY, None) if logging_section is None: raise KeyError(f"The dashboard configuration file {self.filename} did not have a config section {LOGGING_CONFIG_SECTION_KEY}.") logging_section_level = logging_section.get(LOGGING_LEVEL_KEY, None) if logging_section_level is None: raise KeyError(f"The dashboard configuration file {self.filename} did not have a config section {LOGGING_LEVEL_KEY} under section {LOGGING_CONFIG_SECTION_KEY}.") logging_section_file = logging_section.get(LOGGING_FILE_KEY, None) if logging_section_file is None: raise KeyError(f"The dashboard configuration file {self.filename} did not have a config section {LOGGING_FILE_KEY} under section {LOGGING_CONFIG_SECTION_KEY}.") errors_section = self.config_dictionary.get(ERRORS_SECTION_KEY, None) if errors_section is None: raise KeyError(f"The dashboard configuration file {self.filename} did not have a config section {ERRORS_SECTION_KEY}.") contact_name = errors_section.get(CONTACT_NAME_KEY, None) if contact_name is None: raise KeyError( f"The dashboard configuration file {self.filename} did not have a config section {CONTACT_NAME_KEY} under {ERRORS_SECTION_KEY}.") contact_method = errors_section.get(CONTACT_METHOD_KEY, None) if contact_method is None: raise KeyError(f"The dashboard configuration file {self.filename} did not have a config section {CONTACT_METHOD_KEY} under {ERRORS_SECTION_KEY}.") except FileNotFoundError: self.config_dictionary = {}