in src/SAP/SAP_REPORTING/config_validator.py [0:0]
def validate(cfg: dict) -> Union[dict, None]:
"""Validates and processes configuration.
It will discover and log all issues before returning.
Args:
cfg (dict): Config dictionary.
Returns:
dict: Processed config dictionary.
None: In case of validation failure
"""
failed = False
if not cfg.get("deploySAP", False):
logging.info("SAP is not being deployed. Skipping validation.")
return cfg
logging.info("Validating SAP configuration.")
sap = cfg.get("SAP")
if not sap:
logging.error("🛑 Missing 'SAP' values in the config file. 🛑")
return None
missing_sap_attrs = []
for attr in ("deployCDC", "languages", "currencies", "datasets"):
if sap.get(attr) is None or sap.get(attr) == "":
missing_sap_attrs.append(attr)
if missing_sap_attrs:
logging.error("🛑 Missing 'SAP values: %s' in the config file. 🛑",
missing_sap_attrs)
failed = True
datasets = sap.get("datasets")
flavor = sap.get("SQLFlavor", "ecc").lower()
sap["SQLFlavor"] = flavor
cfg_sap_ds_cdc = datasets.get("cdc")
cfg_sap_ds_cdc_ecc = datasets.get("cdcECC")
cfg_sap_ds_cdc_s4 = datasets.get("cdcS4")
if flavor == "ecc":
if cfg_sap_ds_cdc and not cfg_sap_ds_cdc_ecc:
cfg_sap_ds_cdc_ecc = cfg_sap_ds_cdc
else:
cfg_sap_ds_cdc = cfg_sap_ds_cdc_ecc
elif flavor == "s4":
if cfg_sap_ds_cdc and not cfg_sap_ds_cdc_s4:
cfg_sap_ds_cdc_s4 = cfg_sap_ds_cdc
else:
cfg_sap_ds_cdc = cfg_sap_ds_cdc_s4
if not cfg_sap_ds_cdc:
logging.error(("🛑 Cannot resolve SAP/datasets/cdc|cdcECC|cdcS4 values "
"in the config file. 🛑"))
failed = True
datasets["cdc"] = cfg_sap_ds_cdc
datasets["cdcECC"] = cfg_sap_ds_cdc_ecc
datasets["cdcS4"] = cfg_sap_ds_cdc_s4
cfg_sap_ds_raw = datasets.get("raw")
cfg_sap_ds_raw_ecc = datasets.get("rawECC")
cfg_sap_ds_raw_s4 = datasets.get("rawS4")
if flavor == "union":
if not cfg_sap_ds_raw_ecc or not cfg_sap_ds_raw_s4:
logging.error("🛑 SAP/SQLFlavor=union requires "
"all parameters for both ECC and S4. 🛑")
failed = True
elif flavor == "ecc":
if cfg_sap_ds_raw and not cfg_sap_ds_raw_ecc:
cfg_sap_ds_raw_ecc = cfg_sap_ds_raw
else:
cfg_sap_ds_raw = cfg_sap_ds_raw_ecc
elif flavor == "s4":
if cfg_sap_ds_raw and not cfg_sap_ds_raw_s4:
cfg_sap_ds_raw_s4 = cfg_sap_ds_raw
else:
cfg_sap_ds_raw = cfg_sap_ds_raw_s4
if not cfg_sap_ds_raw:
logging.error(("🛑 Cannot resolve SAP/datasets/raw|rawECC|rawS4 values "
"in the config file. 🛑"))
failed = True
datasets["raw"] = cfg_sap_ds_raw
datasets["rawECC"] = cfg_sap_ds_raw_ecc
datasets["rawS4"] = cfg_sap_ds_raw_s4
cfg_sap_ds_reporting = datasets.get("reporting", "REPORTING")
datasets["reporting"] = cfg_sap_ds_reporting
cfg_sap_mandt = sap.get("mandt")
cfg_sap_mandt_ecc = sap.get("mandtECC")
cfg_sap_mandt_s4 = sap.get("mandtS4")
if flavor == "union":
if not cfg_sap_mandt_ecc or not cfg_sap_mandt_s4:
logging.error(("🛑 SAP/SQLFlavor=union requires "
"all parameters for both ECC and S4. 🛑"))
failed = True
elif cfg_sap_mandt_ecc == cfg_sap_mandt_s4:
logging.error(("🛑 Same ECC and S4 MANDT "
"is not allowed for UNION workloads. 🛑"))
failed = True
elif flavor == "ecc":
if cfg_sap_mandt and not cfg_sap_mandt_ecc:
cfg_sap_mandt_ecc = cfg_sap_mandt
else:
cfg_sap_mandt = cfg_sap_mandt_ecc
elif flavor == "s4":
if cfg_sap_mandt and not cfg_sap_mandt_s4:
cfg_sap_mandt_s4 = cfg_sap_mandt
else:
cfg_sap_mandt = cfg_sap_mandt_s4
if not cfg_sap_mandt:
logging.warning("⚠️ Using default SAP Mandt/client = 100.")
cfg_sap_mandt = "100"
sap["mandt"] = cfg_sap_mandt
sap["mandtECC"] = cfg_sap_mandt_ecc or cfg_sap_mandt
sap["mandtS4"] = cfg_sap_mandt_s4 or cfg_sap_mandt
source = cfg["projectIdSource"]
target = cfg["projectIdTarget"]
location = cfg["location"]
datasets = [
resource_validation_helper.DatasetConstraints(
f"{source}.{cfg_sap_ds_raw}", True, cfg.get("testData", False),
location),
resource_validation_helper.DatasetConstraints(
f"{source}.{cfg_sap_ds_cdc}", True, True, location),
resource_validation_helper.DatasetConstraints(
f"{target}.{cfg_sap_ds_reporting}", False, True, location)
]
if not resource_validation_helper.validate_resources([], datasets):
logging.error("🛑 'SAP' resource validation failed. 🛑")
failed = True
if failed:
logging.error("🛑 SAP configuration is invalid. 🛑")
return None
else:
logging.info("✅ SAP configuration is good. ✅")
return cfg