375 lines
17 KiB
Python
375 lines
17 KiB
Python
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 = {}
|