""" Modulo principal para o CI/CD do Data Lake, incluindo empacotamento, implantação e publicação de SQL para Redshift. Este módulo fornece funções para empacotar o projeto, validar e empacotar templates do CloudFormation, implantar stacks do CloudFormation, sincronizar arquivos com o S3 e publicar SQL para Redshift. As funções são projetadas para serem usadas em scripts de CI/CD e automação de tarefas relacionadas ao Data Lake, garantindo que os processos de empacotamento, implantação e publicação sejam realizados de forma confiável e consistente. """ from __future__ import annotations import argparse import json import shutil from pathlib import Path from colorama import Fore from common import get_env_variable, print_message from src.datalake.commons import cloudformation_utils from src.datalake.commons.aws_utils import ( configure_aws_access, delete_files, make_client, make_session, sync_directory_to_s3, ) from src.datalake.commons.ci_utils import VAR_INTERPOLATION from src.datalake.commons.data_env import ( export_to_process, load_runtime_env, set_all_env, ) from src.datalake.commons.version_utils import fix_template_yaml_version from src.interface.python_project import _get_version, create_venv COMMIT_TAG = get_env_variable("CI_COMMIT_TAG") FILE_EXTENSIONS_SELECTION_JSON = "*.json" FILE_EXTENSIONS_SELECTION_SQL = "*.sql" FILE_EXTENSIONS_SELECTION_PY = "*.py" FILE_EXTENSIONS_SELECTION_XML = "*.xml" FILE_EXTENSIONS_SELECTION_XSD = "*.xsd" FILE_EXTENSIONS_SELECTION_WHL = "*.whl" def package(name: str | None) -> dict[str, str]: print_message(f"{Fore.YELLOW}[Dados] Iniciando Package") fix_template_yaml_version() env = load_runtime_env( apply_set_all=True, configure_aws=False, ignore_project_type=False ) if env.get("ENVIRONMENT") != "prod" or env.get("DATAOFFICE_PROJECT_TYPE"): env["CFN_DEPLOYMENT_BUCKET"] = ( f"{env.get('CFN_DEPLOYMENT_BUCKET', '')}-{env.get('ENVIRONMENT', '')}" ) export_to_process(env) cloudformation_utils.cloudformation_validate(env) sync_directory_to_s3( "include", env["WORK_BUCKET_NAME"], f"{env['SOURCE_SYSTEM']}/{env.get('TEAM')}/deploy/include/{env.get('PHASE','all')}/", env, delete=True, ) # Copy src/modules/commons_module to each src/*_fn folder modules_dir = Path("src/modules/commons_module/") if modules_dir.exists(): for fn_dir in Path("src").glob("*_fn"): if fn_dir.is_dir(): print_message( f"[Dados] copiando modulos comuns de: {modules_dir} para {fn_dir}" ) shutil.copytree(modules_dir, fn_dir, dirs_exist_ok=True) final_name = name or env.get("SOURCE_SYSTEM", "") cloudformation_utils.cloudformation_package(env, final_name) return env def _load_deploy_params(env: dict[str, str]) -> list[dict[str, str]]: deploy_file = Path("deploy.json") if not deploy_file.exists(): return [] data = json.loads(deploy_file.read_text(encoding="utf-8")) if not isinstance(data, list): return [] data.extend( ["Environment=${ENVIRONMENT}", "SourceSystem=${SOURCE_SYSTEM}", "Team=${TEAM}"] ) pattern = VAR_INTERPOLATION def sub_vars(s: str) -> str: return pattern.sub(lambda m: env.get(m.group(1), ""), s) params: list[dict[str, str]] = [] for item in data: if not isinstance(item, str) or "=" not in item: continue prepared = item.replace("${ENVIRONMENT}", env.get("ENVIRONMENT", "")) prepared = sub_vars(prepared) key, value = prepared.split("=", 1) params.append({"ParameterKey": key, "ParameterValue": value}) return params def sync_directories_to_s3(env: dict[str, str]) -> None: sync_directory_to_s3( "schema", env["WORK_BUCKET_NAME"], f"{env['SOURCE_SYSTEM']}/schema/", env, delete=True, exclude=["*"], include=[FILE_EXTENSIONS_SELECTION_JSON, FILE_EXTENSIONS_SELECTION_XSD], ) jobs_old_path = Path("jobs") if jobs_old_path.exists(): raise RuntimeError( "The 'jobs' directory is deprecated. Please move your jobs to 'src/jobs'." ) jobs_path = Path("src/jobs") if jobs_path.exists() and jobs_path.is_dir(): for job_dir in jobs_path.iterdir(): if job_dir.is_dir(): sync_directory_to_s3( str(job_dir), env["WORK_BUCKET_NAME"], f"{env['SOURCE_SYSTEM']}/{env.get('TEAM')}/anonymous/glue_jobs/{job_dir.name}/", env, delete=True, exclude=["*"], include=[ FILE_EXTENSIONS_SELECTION_XML, FILE_EXTENSIONS_SELECTION_PY, ], ) print_message("[Dados] Verificando modulos internos") if Path("src/modules").exists(): print_message( "[Dados] Modulos internos encontrados, copiando arquivos .whl para S3" ) sync_directory_to_s3( "dist", env["WORK_BUCKET_NAME"], f"{env['SOURCE_SYSTEM']}/{env.get('TEAM')}/anonymous/glue_jobs/{env.get('PHASE')}/", env, delete=False, exclude=["*"], include=[FILE_EXTENSIONS_SELECTION_WHL], ) sync_directory_to_s3( "column_adjusts/raw", env.get("RAW_BUCKET_NAME", ""), "column_adjusts/raw/", env, delete=True, exclude=["*"], include=[FILE_EXTENSIONS_SELECTION_JSON], ) sync_directory_to_s3( "column_adjusts/stage", env.get("STAGE_BUCKET_NAME", ""), "column_adjusts/stage/", env, delete=True, exclude=["*"], include=[FILE_EXTENSIONS_SELECTION_JSON], ) sync_directory_to_s3( "column_adjusts/analytics", env.get("ANALYTICS_BUCKET_NAME", ""), "column_adjusts/analytics/", env, delete=True, exclude=["*"], include=[FILE_EXTENSIONS_SELECTION_JSON], ) sync_directory_to_s3( "import_files", env.get("ANALYTICS_BUCKET_NAME", ""), "import_files/", env, delete=True, exclude=["*"], include=[FILE_EXTENSIONS_SELECTION_SQL], ) sync_directory_to_s3( "sql/post_consolidate", env["WORK_BUCKET_NAME"], f"{env['SOURCE_SYSTEM']}/{env.get('TEAM')}/sql/post_consolidate/", env, delete=True, exclude=["*"], include=[FILE_EXTENSIONS_SELECTION_SQL], ) sync_directory_to_s3( "dist", env["WORK_BUCKET_NAME"], f"{env['SOURCE_SYSTEM']}/{env.get('TEAM')}/glue_jobs/", env, delete=False, exclude=["*"], include=[FILE_EXTENSIONS_SELECTION_WHL], ) if COMMIT_TAG: print_message( "[Dados] python libs - versão fechada, removendo antigos dev/snapshot..." ) delete_files( env["WORK_BUCKET_NAME"], f"{env['SOURCE_SYSTEM']}/{env.get('TEAM')}/glue_jobs/", env, target_name="SNAPSHOT", target_extension=".whl", ) delete_files( env["WORK_BUCKET_NAME"], f"{env['SOURCE_SYSTEM']}/{env.get('TEAM')}/glue_jobs/", env, target_name=".dev", target_extension=".whl", ) else: print_message( "[Dados] python libs - em desenvolvimento, não removendo antigos dev/snapshot..." ) def deploy(name: str | None) -> None: print_message(f"{Fore.YELLOW}[Dados] Preparando Deploy") env = package(name) env = configure_aws_access(env) env = set_all_env(env, force=True) env["STATE_MACHINE_NAME"] = f"{env['ENVIRONMENT']}-{env['SOURCE_SYSTEM']}-pipeline" env["THIRD_PARTY_STATE_MACHINE_NAME"] = ( f"{env['ENVIRONMENT']}-{env['SOURCE_SYSTEM']}-{env['TEAM']}-pipeline" ) env["TRANSIENT_STATE_MACHINE_NAME"] = ( f"{env['ENVIRONMENT']}-{env['SOURCE_SYSTEM']}-transient-pipeline" ) env["MODULE_VERSION"] = _get_version() export_to_process(env) print_message(f"{Fore.YELLOW}[Dados] Iniciando Deploy") session = make_session(env) cf = make_client(session, "cloudformation", region_name=env.get("REGION", "")) stack_name = env.get("STACK_NAME", "") template_body = Path("packaged.yaml").read_text(encoding="utf-8") parameters = _load_deploy_params(env) tags = [ {"Key": "env", "Value": env.get("ENVIRONMENT", "")}, {"Key": "base", "Value": name or env.get("SOURCE_SYSTEM", "")}, {"Key": "project", "Value": Path.cwd().name}, {"Key": "phase", "Value": "deploy"}, {"Key": "type", "Value": "all"}, {"Key": "team", "Value": env.get("TEAM", "")}, ] print_message( f"{Fore.YELLOW}[Dados] Deploy - stack: {stack_name} - template_body size: {len(template_body)} " f"\n- parameters: {json.dumps(parameters, ensure_ascii=False, indent=2)} " f"\n- tags: {json.dumps(tags, ensure_ascii=False, indent=2)}" ) try: print_message(f"{Fore.YELLOW}[Dados] Deploy - describe_stacks") cf.describe_stacks(StackName=stack_name) print_message(f"{Fore.YELLOW}[Dados] Deploy - update_stack") cf.update_stack( StackName=stack_name, TemplateBody=template_body, Capabilities=[ "CAPABILITY_IAM", "CAPABILITY_AUTO_EXPAND", "CAPABILITY_NAMED_IAM", ], Parameters=parameters, Tags=tags, ) print_message(f"{Fore.YELLOW}[Dados] Deploy - get_waiter") waiter = cf.get_waiter("stack_update_complete") waiter.wait(StackName=stack_name) except Exception as exc: message = str(exc) if "No updates are to be performed" not in message: response = cf.describe_stacks(StackName=stack_name) stack = response["Stacks"][0] error_reason = stack.get("StackStatusReason", "No reason provided.") print_message( f"CloudFormation Error: {error_reason} - response: {response}" ) stack_events = cf.describe_stack_events(StackName=stack_name) print_message("Recent CloudFormation Stack Events:") # Imprime os N últimos eventos do stack, limitando a 20 para evitar excesso de logs for event in stack_events.get("StackEvents", [])[:20]: print_message( f" - {event['Timestamp']}: {event['ResourceStatus']} " f"- {event['ResourceType']} - {event.get('ResourceStatusReason', '')}" ) raise print_message( f"{Fore.YELLOW}[Dados] Deploy finalizado - {stack_name} - Inciando copia de arquivos para S3" ) sync_directories_to_s3(env) def main() -> int: parser = argparse.ArgumentParser(description="Data CI main") parser.add_argument("-d", dest="deploy_name", nargs="?", const="") args = parser.parse_args() try: if args.deploy_name is not None: print_message("[Dados] Criando o ambiente virtual") create_venv() deploy(args.deploy_name or None) else: parser.print_help() return 2 except Exception as exc: print_message(str(exc)) return 2 return 0 if __name__ == "__main__": raise SystemExit(main())