def build_notification_body()

in src/backend/entrypoints/lambdas/lambda_run.py [0:0]

• 30 lines of code
• 9 McCabe index (conditional complexity)

def build_notification_body(lambda_event: Dict[str, Any], lambda_session: Dict[str, Any]) -> str:
    alert_text_template = lambda_event["alert_text_template"]
    alert_text_template = re.sub(r"{\s*([^}]+)\s*}", lambda match: "{" + match.group(1).strip() + "}", alert_text_template)

    template_keys_and_columns = re.findall(r"{([^}\[\]]+)(\[[^]]+])?}", alert_text_template)

    def get_template_value(session_key: str, template_key: str, column: str, lambda_session: Dict[str, Any]) -> float:
        value = lambda_session[session_key][template_key]
        if isinstance(value, pd.DataFrame):
            try:
                value = value[column].iloc[-1]
            except KeyError:
                value = None
        elif isinstance(value, pd.Series):
            value = value.iloc[-1]
        return value

    mappings = {}
    for template_key, column in template_keys_and_columns:
        allowed_key = f"{template_key}_{column[2:-2]}" if column else template_key
        mappings[allowed_key] = next(
            (
                get_template_value(session_key, template_key, "close" if column == "" else column[2:-2], lambda_session)
                for session_key in ["data_by_symbol", "data_by_synth", "data_by_indicator"]
                if template_key in lambda_session[session_key]
            ),
            None,
        )

    template_with_allowed_keys = re.sub(
        r"(\w+)\['(\w+)']", lambda match: match.group(1) + "_" + match.group(2), alert_text_template
    )

    alert_text = template_with_allowed_keys.format(**mappings)
    return alert_text