""" Utilitários para interagir com o AWS Lambda, incluindo invocação de funções e tratamento de payloads. Este módulo fornece funções para invocar funções Lambda, resolver ARNs de funções a partir de nomes e gerenciar payloads de entrada e saída. As funções são projetadas para serem usadas em scripts de CI/CD e automação de tarefas relacionadas à AWS Lambda, garantindo que as invocações sejam realizadas de forma confiável e que os resultados sejam validados conforme necessário. """ from __future__ import annotations import base64 import json from pathlib import Path from typing import Any from botocore.config import Config from botocore.exceptions import ReadTimeoutError from common import print_message from src.datalake.commons.aws_utils import ( make_client, make_session, resolve_stack_output, ) def _resolve_lambda_arn(name: str, env: dict[str, str]) -> str: """ Resolve o ARN de uma função Lambda a partir do nome ou ARN fornecido. Args: name (str): Nome da função Lambda ou ARN completo. env (dict[str, str]): Dicionário contendo as variáveis de ambiente. Returns: str: ARN completo da função Lambda. """ if name.startswith("arn:"): print_message(f"[Dados] lambda - invoke - arn: {name}") return name stack_name = env.get("STACK_NAME", "") region = env.get("REGION", "") if not stack_name or not region: raise RuntimeError( "[Dados] STACK_NAME e REGION são obrigatórios para resolver ARN de output" ) session = make_session(env) arn = resolve_stack_output( session, stack_name=stack_name, output_key=name, region_name=region, ) print_message(f"[Dados] lambda - invoke - ID - arn: {arn}") if not arn: raise RuntimeError( f"[Dados] lambda - invoke - Não encontrou - '{name}' em '{stack_name}' verifique o template.yaml!" ) return arn def load_request_payload(payload: str | None) -> str: """ Carrega o payload de solicitação para invocação da função Lambda. Args: payload (str | None): Caminho para o arquivo de payload ou string JSON. Returns: str: Conteúdo do payload como string JSON. """ request_payload = "{}" if payload: if isinstance(payload, str) and payload.startswith("file://"): payload_file = Path(payload[len("file://") :]) request_payload = payload_file.read_text(encoding="utf-8") elif isinstance(payload, (bytes, bytearray)): request_payload = payload.decode("utf-8") else: request_payload = payload return request_payload def load_log_result_text(response) -> str: """ Carrega o resultado do log da resposta da invocação da função Lambda. Args: response: Resposta da invocação da função Lambda. Returns: str: Texto do log da resposta da invocação da função Lambda. """ log_result_payload = response.get("LogResult", "") if isinstance(log_result_payload, bytes): log_result_text: str = log_result_payload.decode("utf-8") else: try: log_result_text: str = base64.b64decode(log_result_payload).decode("utf-8") except Exception as exc: print_message(str(exc)) log_result_text: str = str(log_result_payload) if "timeout" in log_result_text.lower(): print_message( "[Dados] lambda - invoke response - Timeout detectado, considere aumentar o timeout da função Lambda" ) else: print_message( "[Dados] lambda - invoke response - Lambda executou dentro do tempo limite", debug_message=True, ) return log_result_text def invoke_lambda( name: str, payload: str | None, env: dict[str, str], validate: bool = True, expect_success: bool = False, ignore_timeout: bool = False, description: str | None = None, ) -> dict[str, Any] | None: """ Invoca uma função Lambda com o payload fornecido e opções de validação. Args: name (str): Nome da função Lambda. payload (str | None): Caminho para o arquivo de payload ou string JSON. env (dict[str, str]): Dicionário contendo as variáveis de ambiente. validate (bool): Indica se a resposta deve ser validada. expect_success (bool): Indica se o sucesso é esperado na resposta. ignore_timeout (bool): Indica se o timeout deve ser ignorado. description (str | None): Descrição da invocação. Returns: dict[str, Any] | None Raises: ValueError: Se ocorrer um erro na execução da função Lambda ou se a validação falhar. KeyError: Se a resposta não contiver a chave 'success' quando o sucesso for esperado. """ arn = _resolve_lambda_arn(name, env) request_payload = load_request_payload(payload) session = make_session(env) timeout_config = Config( connect_timeout=120, read_timeout=905, retries={"max_attempts": 0, "mode": "standard"}, ) lambda_client = make_client( session, "lambda", region_name=env.get("REGION", ""), timeout_config=timeout_config, ) try: response = lambda_client.invoke( FunctionName=arn, InvocationType="RequestResponse", LogType="Tail", Payload=request_payload, ) print_message( f"[Dados] lambda - invoke response - {name} - {description or ''} - \n{response}", debug_message=True, ) response_payload_dict = load_response_payload_dict(response) log_result_text = load_log_result_text(response) response_json = { "StatusCode": response.get("StatusCode"), "FunctionError": response.get("FunctionError"), "ExecutedVersion": response.get("ExecutedVersion"), "LogResult": log_result_text, "ResponsePayload": response_payload_dict, } print_message( f"[Dados] lambda - invoke response_json - {name} - {description or ''} " f"- \n{json.dumps(response_json, indent=2)}", debug_message=True, ) validate_response(name, description, validate, response_json) if ( expect_success and response_payload_dict and not response_payload_dict.get("success", False) ): raise KeyError( "[Dados] lambda - Erro de execução encontrado, invalidando! " "'success' não encontrado no payload de resposta" ) return response_json except ReadTimeoutError: if ( ignore_timeout or env.get("SOPS_DATACORE_LAMBDA_IGNORE_TIME_OUT", "false").lower() == "true" ): print_message( "[Dados] lambda - invoke response - Timeout detectado, mas ignorando devido à configuração de ambiente." ) return {} raise def load_response_payload_dict(response) -> dict[Any, Any]: """ Carrega o payload de resposta da invocação da função Lambda. Args: response: Resposta da invocação da função Lambda. Returns: dict[Any, Any]: Dicionário contendo o payload de resposta da função Lambda. """ response_payload = response.get("Payload") payload_text = response_payload.read().decode("utf-8") if response_payload else "" if payload_text: return json.loads(payload_text) return {} def validate_response( name: str, description: str | None, validate: bool, response_json: dict[str, Any] ) -> None: """ Valida a resposta da invocação da função Lambda. Args: name (str): Nome da função Lambda. description (str | None): Descrição da invocação. validate (bool): Indica se a resposta deve ser validada. response_json (dict[str, Any]): Dicionário contendo a resposta da função Lambda. Returns: None """ if validate: if ( "FunctionError" in response_json and response_json["FunctionError"] or "errorMessage" in response_json.get("ResponsePayload", {}) and response_json["ResponsePayload"]["errorMessage"] ): print_message( f"[Dados] lambda - {name} - {description or ''} " f"- Detalhes do erro: {json.dumps(response_json, indent=2)}" ) raise ValueError( "[Dados] lambda - Erro de execução encontrado, invalidando job!" " Erro: 'FunctionError' ou 'errorMessage'" ) print_message(f"[Dados] lambda - {name} - {description or ''} - Tudo feito!")