当前位置：首页 > news >正文

AWS EC2 Jenkins 自动化部署与 Python 环境管理

news 2025/10/17 6:09:55

概述

本文将详细介绍在 AWS 上创建和管理 EC2 实例集群，配置 Jenkins 持续集成环境，以及使用 Poetry 管理 Python 项目的完整流程。我们将通过 Python 脚本实现自动化配置和故障检查。

AWS上创建EC2的Linux实例，选择一个AMI，创建SSH Key，允许终端连接，并安装配置Jenkins软件和Python 3.12，使用poetry管理Python包，可以安装git插件来获取指定分支最新版本的所有代码，其中包含配置信息文件pyproject.toml，job运行时，可以切换到配置信息文件所在的目录，用poetry install --no-root安装需要的包，再用poetry run python命令行运行python代码文件，并创建多个EC2实例，运行在job配置信息中选择EC2实例运行Python代码。允许EC2连接到指定的S3存储桶和Redshift实例，通过VPC、安全组和防火墙实现网络路由和网络安全，通过Route53实现DNS服务，用CloudWatch实现监控。

本文详细介绍了在 AWS 上创建和管理 Jenkins EC2 实例集群的完整流程，包括：

基础设施自动化: 使用 Python 脚本自动创建 VPC、安全组、EC2 实例等
Jenkins 配置: 自动安装和配置 Jenkins 及必要插件
Python 环境管理: 使用 Poetry 管理 Python 依赖和项目配置
网络和安全: 通过安全组、IAM 角色实现安全的网络访问
监控和告警: 使用 CloudWatch 进行系统监控
故障恢复: 实现自动化的健康检查和实例恢复

这种架构提供了高度可扩展和可靠的持续集成环境，能够满足企业级的 Python 项目部署需求。通过自动化脚本，大大减少了手动操作，提高了部署的可靠性和一致性。

架构设计

网络架构

VPC: 创建私有网络环境
子网: 公有和私有子网分离
安全组: 精细的访问控制
Route53: DNS 管理和服务发现
Internet Gateway/NAT Gateway: 网络出口配置

计算架构

EC2 实例: Jenkins master 和多个 worker 节点
Auto Scaling: 根据负载自动调整实例数量
负载均衡: 分布式流量管理

存储与数据

S3: 代码存储和构建产物
Redshift: 数据仓库和分析
EBS: 实例持久化存储

实现步骤

1. 基础设施自动化配置

#!/usr/bin/env python3
"""
AWS EC2 Jenkins 集群自动化部署脚本
支持自动配置、故障检查和恢复
"""import boto3
import json
import time
import subprocess
import paramiko
from botocore.exceptions import ClientError
import logging
import sys
from pathlib import Pathclass AWSJenkinsDeployer:def __init__(self, region='us-east-1'):self.region = regionself.ec2 = boto3.resource('ec2', region_name=region)self.ec2_client = boto3.client('ec2', region_name=region)self.iam = boto3.client('iam', region_name=region)self.s3 = boto3.client('s3', region_name=region)self.route53 = boto3.client('route53', region_name=region)self.cloudwatch = boto3.client('cloudwatch', region_name=region)# 配置日志logging.basicConfig(level=logging.INFO,format='%(asctime)s - %(levelname)s - %(message)s')self.logger = logging.getLogger(__name__)# 配置参数self.config = self.load_config()def load_config(self):"""加载配置文件"""config = {'vpc_name': 'jenkins-vpc','vpc_cidr': '10.0.0.0/16','public_subnet_cidr': '10.0.1.0/24','private_subnet_cidr': '10.0.2.0/24','jenkins_instance_type': 't3.medium','worker_instance_type': 't3.large','key_name': 'jenkins-key','jenkins_ami': 'ami-0c02fb55956c7d316',  # Amazon Linux 2's3_bucket': 'jenkins-artifacts-2024','domain_name': 'jenkins.yourdomain.com','redshift_cluster_identifier': 'jenkins-redshift'}return configdef create_vpc_and_networking(self):"""创建 VPC 和网络组件"""try:# 创建 VPCvpc = self.ec2.create_vpc(CidrBlock=self.config['vpc_cidr'])vpc.create_tags(Tags=[{'Key': 'Name', 'Value': self.config['vpc_name']}])vpc.wait_until_available()# 创建互联网网关igw = self.ec2.create_internet_gateway()vpc.attach_internet_gateway(InternetGatewayId=igw.id)# 创建路由表public_route_table = vpc.create_route_table()public_route_table.create_route(DestinationCidrBlock='0.0.0.0/0',GatewayId=igw.id)# 创建子网public_subnet = vpc.create_subnet(CidrBlock=self.config['public_subnet_cidr'])private_subnet = vpc.create_subnet(CidrBlock=self.config['private_subnet_cidr'])# 关联路由表public_route_table.associate_with_subnet(SubnetId=public_subnet.id)self.logger.info("VPC 和网络组件创建成功")return {'vpc_id': vpc.id,'public_subnet_id': public_subnet.id,'private_subnet_id': private_subnet.id}except ClientError as e:self.logger.error(f"创建 VPC 失败: {e}")raisedef create_security_groups(self, vpc_id):"""创建安全组"""try:# Jenkins 安全组jenkins_sg = self.ec2.create_security_group(GroupName='jenkins-sg',Description='Security group for Jenkins',VpcId=vpc_id)# 添加入站规则jenkins_sg.authorize_ingress(IpPermissions=[{'IpProtocol': 'tcp','FromPort': 22,'ToPort': 22,'IpRanges': [{'CidrIp': '0.0.0.0/0'}]},{'IpProtocol': 'tcp','FromPort': 8080,'ToPort': 8080,'IpRanges': [{'CidrIp': '0.0.0.0/0'}]},{'IpProtocol': 'tcp','FromPort': 50000,'ToPort': 50000,'IpRanges': [{'CidrIp': '0.0.0.0/0'}]}])# Worker 节点安全组worker_sg = self.ec2.create_security_group(GroupName='worker-sg',Description='Security group for worker nodes',VpcId=vpc_id)worker_sg.authorize_ingress(IpPermissions=[{'IpProtocol': 'tcp','FromPort': 22,'ToPort': 22,'IpRanges': [{'CidrIp': '10.0.0.0/16'}]}])self.logger.info("安全组创建成功")return {'jenkins_sg_id': jenkins_sg.id,'worker_sg_id': worker_sg.id}except ClientError as e:self.logger.error(f"创建安全组失败: {e}")raisedef create_iam_roles(self):"""创建 IAM 角色和策略"""try:# 创建 EC2 实例角色assume_role_policy = {"Version": "2012-10-17","Statement": [{"Effect": "Allow","Principal": {"Service": "ec2.amazonaws.com"},"Action": "sts:AssumeRole"}]}role = self.iam.create_role(RoleName='JenkinsEC2Role',AssumeRolePolicyDocument=json.dumps(assume_role_policy),Description='Role for Jenkins EC2 instances')# 附加策略s3_policy = {"Version": "2012-10-17","Statement": [{"Effect": "Allow","Action": ["s3:GetObject","s3:PutObject","s3:ListBucket"],"Resource": [f"arn:aws:s3:::{self.config['s3_bucket']}",f"arn:aws:s3:::{self.config['s3_bucket']}/*"]}]}self.iam.put_role_policy(RoleName='JenkinsEC2Role',PolicyName='S3Access',PolicyDocument=json.dumps(s3_policy))# 创建实例配置文件instance_profile = self.iam.create_instance_profile(InstanceProfileName='JenkinsInstanceProfile')self.iam.add_role_to_instance_profile(InstanceProfileName='JenkinsInstanceProfile',RoleName='JenkinsEC2Role')self.logger.info("IAM 角色创建成功")return instance_profile['InstanceProfile']['Arn']except ClientError as e:self.logger.error(f"创建 IAM 角色失败: {e}")raisedef create_key_pair(self):"""创建 SSH 密钥对"""try:key_pair = self.ec2_client.create_key_pair(KeyName=self.config['key_name'])# 保存私钥到文件with open(f"{self.config['key_name']}.pem", 'w') as f:f.write(key_pair['KeyMaterial'])# 设置权限subprocess.run(['chmod', '400', f"{self.config['key_name']}.pem"])self.logger.info("SSH 密钥对创建成功")return Trueexcept ClientError as e:self.logger.error(f"创建密钥对失败: {e}")raisedef create_jenkins_instance(self, subnet_id, sg_id, instance_profile_arn):"""创建 Jenkins 主实例"""try:user_data = """#!/bin/bash# 更新系统yum update -y# 安装 Javaamazon-linux-extras install java-openjdk11 -y# 安装 Jenkinswget -O /etc/yum.repos.d/jenkins.repo https://pkg.jenkins.io/redhat-stable/jenkins.reporpm --import https://pkg.jenkins.io/redhat-stable/jenkins.io.keyyum install jenkins -y# 安装 Python 3.12 和 Poetryyum install python3.12 python3.12-pip -ypip3.12 install poetry# 安装 Gityum install git -y# 启动 Jenkinssystemctl enable jenkinssystemctl start jenkins# 安装 AWS CLIcurl "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip"yum install unzip -yunzip awscliv2.zip./aws/installecho "Jenkins installation completed""""instances = self.ec2.create_instances(ImageId=self.config['jenkins_ami'],MinCount=1,MaxCount=1,InstanceType=self.config['jenkins_instance_type'],KeyName=self.config['key_name'],NetworkInterfaces=[{'AssociatePublicIpAddress': True,'DeviceIndex': 0,'SubnetId': subnet_id,'Groups': [sg_id]}],IamInstanceProfile={'Arn': instance_profile_arn},UserData=user_data,TagSpecifications=[{'ResourceType': 'instance','Tags': [{'Key': 'Name', 'Value': 'Jenkins-Master'}]}])instance = instances[0]instance.wait_until_running()instance.reload()self.logger.info(f"Jenkins 实例创建成功: {instance.id}")return instanceexcept ClientError as e:self.logger.error(f"创建 Jenkins 实例失败: {e}")raisedef create_worker_instances(self, subnet_id, sg_id, count=2):"""创建工作节点实例"""try:user_data = """#!/bin/bash# 更新系统yum update -y# 安装 Javaamazon-linux-extras install java-openjdk11 -y# 安装 Python 3.12 和 Poetryyum install python3.12 python3.12-pip -ypip3.12 install poetry# 安装 Gityum install git -y# 安装 AWS CLIcurl "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip"yum install unzip -yunzip awscliv2.zip./aws/installecho "Worker node setup completed""""instances = self.ec2.create_instances(ImageId=self.config['jenkins_ami'],MinCount=count,MaxCount=count,InstanceType=self.config['worker_instance_type'],KeyName=self.config['key_name'],NetworkInterfaces=[{'AssociatePublicIpAddress': True,'DeviceIndex': 0,'SubnetId': subnet_id,'Groups': [sg_id]}],UserData=user_data,TagSpecifications=[{'ResourceType': 'instance','Tags': [{'Key': 'Name', 'Value': 'Jenkins-Worker'}]}])for instance in instances:instance.wait_until_running()instance.reload()self.logger.info(f"创建了 {count} 个工作节点")return instancesexcept ClientError as e:self.logger.error(f"创建工作节点失败: {e}")raisedef setup_jenkins_via_ssh(self, instance):"""通过 SSH 配置 Jenkins"""try:# 等待实例完全启动time.sleep(120)key = paramiko.RSAKey.from_private_key_file(f"{self.config['key_name']}.pem")ssh = paramiko.SSHClient()ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())ssh.connect(instance.public_ip_address, username='ec2-user', pkey=key)# 获取 Jenkins 初始密码stdin, stdout, stderr = ssh.exec_command('sudo cat /var/lib/jenkins/secrets/initialAdminPassword')jenkins_password = stdout.read().decode().strip()# 安装 Jenkins 插件plugins_script = """# 等待 Jenkins 启动until sudo systemctl is-active jenkins | grep -q "active"; dosleep 10done# 安装必要插件sudo wget -O /var/lib/jenkins/plugins/git.hpi \\https://updates.jenkins.io/download/plugins/git/latest/git.hpisudo wget -O /var/lib/jenkins/plugins/aws-credentials.hpi \\https://updates.jenkins.io/download/plugins/aws-credentials/latest/aws-credentials.hpisudo wget -O /var/lib/jenkins/plugins/ec2.hpi \\https://updates.jenkins.io/download/plugins/ec2/latest/ec2.hpi# 重启 Jenkins 加载插件sudo systemctl restart jenkins"""stdin, stdout, stderr = ssh.exec_command(plugins_script)stdout.channel.recv_exit_status()ssh.close()self.logger.info(f"Jenkins 初始密码: {jenkins_password}")self.logger.info(f"Jenkins URL: http://{instance.public_ip_address}:8080")return jenkins_passwordexcept Exception as e:self.logger.error(f"SSH 配置失败: {e}")raisedef create_s3_bucket(self):"""创建 S3 存储桶"""try:self.s3.create_bucket(Bucket=self.config['s3_bucket'],CreateBucketConfiguration={'LocationConstraint': self.region})# 配置存储桶策略bucket_policy = {"Version": "2012-10-17","Statement": [{"Effect": "Allow","Principal": {"AWS": "*"},"Action": ["s3:GetObject","s3:PutObject"],"Resource": f"arn:aws:s3:::{self.config['s3_bucket']}/*"}]}self.s3.put_bucket_policy(Bucket=self.config['s3_bucket'],Policy=json.dumps(bucket_policy))self.logger.info(f"S3 存储桶创建成功: {self.config['s3_bucket']}")except ClientError as e:self.logger.error(f"创建 S3 存储桶失败: {e}")raisedef setup_cloudwatch_alarms(self, instance_ids):"""设置 CloudWatch 告警"""try:for instance_id in instance_ids:# CPU 使用率告警self.cloudwatch.put_metric_alarm(AlarmName=f'HighCPU-{instance_id}',AlarmDescription='Alarm when CPU exceeds 80%',MetricName='CPUUtilization',Namespace='AWS/EC2',Statistic='Average',Dimensions=[{'Name': 'InstanceId', 'Value': instance_id}],Period=300,EvaluationPeriods=2,Threshold=80.0,ComparisonOperator='GreaterThanThreshold',AlarmActions=[],  # 可以添加 SNS 主题OKActions=[])# 状态检查告警self.cloudwatch.put_metric_alarm(AlarmName=f'StatusCheckFailed-{instance_id}',AlarmDescription='Alarm when status check fails',MetricName='StatusCheckFailed',Namespace='AWS/EC2',Statistic='Maximum',Dimensions=[{'Name': 'InstanceId', 'Value': instance_id}],Period=300,EvaluationPeriods=1,Threshold=1.0,ComparisonOperator='GreaterThanOrEqualToThreshold',AlarmActions=[],OKActions=[])self.logger.info("CloudWatch 告警设置成功")except ClientError as e:self.logger.error(f"设置 CloudWatch 告警失败: {e}")raisedef deploy_infrastructure(self):"""部署完整基础设施"""self.logger.info("开始部署基础设施...")# 创建网络组件networking = self.create_vpc_and_networking()# 创建安全组security_groups = self.create_security_groups(networking['vpc_id'])# 创建 IAM 角色instance_profile_arn = self.create_iam_roles()# 创建密钥对self.create_key_pair()# 创建 S3 存储桶self.create_s3_bucket()# 创建 Jenkins 实例jenkins_instance = self.create_jenkins_instance(networking['public_subnet_id'],security_groups['jenkins_sg_id'],instance_profile_arn)# 创建工作节点worker_instances = self.create_worker_instances(networking['public_subnet_id'],security_groups['worker_sg_id'])# 配置 Jenkinsjenkins_password = self.setup_jenkins_via_ssh(jenkins_instance)# 设置监控all_instance_ids = [jenkins_instance.id] + [inst.id for inst in worker_instances]self.setup_cloudwatch_alarms(all_instance_ids)self.logger.info("基础设施部署完成!")return {'jenkins_instance': jenkins_instance,'worker_instances': worker_instances,'jenkins_password': jenkins_password,'networking': networking}class JenkinsJobManager:"""Jenkins 任务管理器"""def __init__(self, jenkins_url, username, password):self.jenkins_url = jenkins_urlself.username = usernameself.password = passworddef create_python_pipeline_job(self, job_name, git_repo, branch, python_script):"""创建 Python 流水线任务"""pipeline_script = f"""
pipeline {{agent anyenvironment {{PYTHON_VERSION = "3.12"POETRY_VERSION = "1.7.1"}}stages {{stage('Checkout') {{steps {{git branch: '{branch}',url: '{git_repo}'}}}}stage('Setup Python Environment') {{steps {{sh '''python3.12 -m pip install --upgrade pippython3.12 -m pip install poetrypoetry install --no-root'''}}}}stage('Run Python Script') {{steps {{sh '''cd ${{WORKSPACE}}poetry run python {python_script}'''}}}}stage('Upload Artifacts to S3') {{steps {{sh '''aws s3 sync ./output/ s3://your-bucket/jenkins-artifacts/${{JOB_NAME}}/${{BUILD_NUMBER}}/'''}}}}}}post {{always {{publishHTML([allowMissing: false,alwaysLinkToLastBuild: false,keepAll: true,reportDir: 'reports',reportFiles: '*.html',reportName: 'Test Report'])}}success {{emailext (subject: "SUCCESS: Job '${{JOB_NAME}} - ${{BUILD_NUMBER}}'",body: "The job ${{JOB_NAME}} completed successfully.",to: "devops@yourcompany.com")}}failure {{emailext (subject: "FAILED: Job '${{JOB_NAME}} - ${{BUILD_NUMBER}}'",body: "The job ${{JOB_NAME}} failed.",to: "devops@yourcompany.com")}}}}
}}
"""return pipeline_scriptclass HealthChecker:"""系统健康检查器"""def __init__(self, deployer):self.deployer = deployerdef check_ec2_instances(self):"""检查 EC2 实例状态"""try:instances = self.deployer.ec2.instances.all()healthy_instances = []unhealthy_instances = []for instance in instances:if instance.state['Name'] == 'running':# 执行系统状态检查try:response = self.deployer.ec2_client.describe_instance_status(InstanceIds=[instance.id],IncludeAllInstances=True)status = response['InstanceStatuses'][0]system_status = status['SystemStatus']['Status']instance_status = status['InstanceStatus']['Status']if system_status == 'ok' and instance_status == 'ok':healthy_instances.append(instance.id)else:unhealthy_instances.append(instance.id)except ClientError:unhealthy_instances.append(instance.id)return {'healthy': healthy_instances,'unhealthy': unhealthy_instances}except ClientError as e:self.deployer.logger.error(f"检查 EC2 实例状态失败: {e}")return {'healthy': [], 'unhealthy': []}def check_jenkins_service(self, instance_ip):"""检查 Jenkins 服务状态"""try:key = paramiko.RSAKey.from_private_key_file(f"{self.deployer.config['key_name']}.pem")ssh = paramiko.SSHClient()ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())ssh.connect(instance_ip, username='ec2-user', pkey=key)stdin, stdout, stderr = ssh.exec_command('sudo systemctl is-active jenkins')status = stdout.read().decode().strip()ssh.close()return status == 'active'except Exception as e:self.deployer.logger.error(f"检查 Jenkins 服务失败: {e}")return Falsedef check_s3_connectivity(self):"""检查 S3 连接性"""try:self.deployer.s3.list_buckets()return Trueexcept ClientError as e:self.deployer.logger.error(f"S3 连接检查失败: {e}")return Falsedef comprehensive_health_check(self, jenkins_instance):"""综合健康检查"""self.deployer.logger.info("执行综合健康检查...")results = {'ec2_instances': self.check_ec2_instances(),'jenkins_service': self.check_jenkins_service(jenkins_instance.public_ip_address),'s3_connectivity': self.check_s3_connectivity()}# 输出检查结果healthy_count = len(results['ec2_instances']['healthy'])unhealthy_count = len(results['ec2_instances']['unhealthy'])self.deployer.logger.info(f"EC2 实例状态: {healthy_count} 健康, {unhealthy_count} 不健康")self.deployer.logger.info(f"Jenkins 服务: {'运行正常' if results['jenkins_service'] else '异常'}")self.deployer.logger.info(f"S3 连接性: {'正常' if results['s3_connectivity'] else '异常'}")return resultsdef main():"""主函数"""try:# 初始化部署器deployer = AWSJenkinsDeployer()# 部署基础设施deployment = deployer.deploy_infrastructure()# 执行健康检查health_checker = HealthChecker(deployer)health_results = health_checker.comprehensive_health_check(deployment['jenkins_instance'])# 输出部署信息print("\n" + "="*50)print("部署完成!")print("="*50)print(f"Jenkins URL: http://{deployment['jenkins_instance'].public_ip_address}:8080")print(f"初始密码: {deployment['jenkins_password']}")print(f"工作节点数量: {len(deployment['worker_instances'])}")print(f"S3 存储桶: {deployer.config['s3_bucket']}")except Exception as e:logging.error(f"部署失败: {e}")sys.exit(1)if __name__ == "__main__":main()

故障检查和自动化恢复

1. 自动化监控脚本

"""
自动化故障检查和恢复脚本
"""class AutoHealer:def __init__(self, deployer):self.deployer = deployerself.health_checker = HealthChecker(deployer)def auto_heal_instances(self):"""自动修复不健康的实例"""health_status = self.health_checker.check_ec2_instances()for instance_id in health_status['unhealthy']:self.deployer.logger.warning(f"检测到不健康实例: {instance_id}")try:# 尝试重启实例self.deployer.ec2_client.reboot_instances(InstanceIds=[instance_id])self.deployer.logger.info(f"已重启实例: {instance_id}")except ClientError as e:self.deployer.logger.error(f"重启实例 {instance_id} 失败: {e}")# 如果重启失败，终止并创建新实例self.replace_instance(instance_id)def replace_instance(self, instance_id):"""替换故障实例"""try:# 获取实例信息instance = self.deployer.ec2.Instance(instance_id)# 终止故障实例instance.terminate()instance.wait_until_terminated()self.deployer.logger.info(f"已终止故障实例: {instance_id}")# 创建新实例替换# 这里可以根据需要实现具体的替换逻辑except ClientError as e:self.deployer.logger.error(f"替换实例 {instance_id} 失败: {e}")

2. 定期检查脚本

"""
定期健康检查脚本
"""def schedule_health_checks():"""调度定期健康检查"""import scheduleimport timedeployer = AWSJenkinsDeployer()health_checker = HealthChecker(deployer)healer = AutoHealer(deployer)def daily_health_check():deployer.logger.info("执行每日健康检查...")# 获取所有 Jenkins 相关实例instances = deployer.ec2.instances.filter(Filters=[{'Name': 'tag:Name', 'Values': ['Jenkins-Master', 'Jenkins-Worker']}])jenkins_instances = [inst for inst in instances if inst.state['Name'] == 'running']if jenkins_instances:health_results = health_checker.comprehensive_health_check(jenkins_instances[0])# 如果有不健康的实例，尝试自动修复if health_results['ec2_instances']['unhealthy']:healer.auto_heal_instances()# 每天执行健康检查schedule.every().day.at("06:00").do(daily_health_check)# 每5分钟检查一次 Jenkins 服务schedule.every(5).minutes.do(lambda: health_checker.check_jenkins_service('your-jenkins-ip'))while True:schedule.run_pending()time.sleep(60)

配置管理和最佳实践

1. Poetry 配置示例 (pyproject.toml)

[tool.poetry]
name = "jenkins-python-project"
version = "0.1.0"
description = "Python project for Jenkins automation"
authors = ["Your Name <your.email@example.com>"][tool.poetry.dependencies]
python = "^3.12"
boto3 = "^1.34.0"
paramiko = "^3.3.0"
pandas = "^2.0.0"
sqlalchemy = "^2.0.0"
psycopg2-binary = "^2.9.0"[tool.poetry.group.dev.dependencies]
pytest = "^7.0.0"
black = "^23.0.0"
flake8 = "^6.0.0"[build-system]
requires = ["poetry-core"]
build-backend = "poetry.core.masonry.api"[tool.poetry.scripts]
deploy = "deploy_script:main"
health-check = "health_checker:main"

2. Jenkinsfile 示例

pipeline {agent {label 'python-3.12'}environment {AWS_DEFAULT_REGION = 'us-east-1'S3_BUCKET = 'jenkins-artifacts-2024'REDSHIFT_HOST = 'your-redshift-cluster.region.redshift.amazonaws.com'}stages {stage('Checkout') {steps {git branch: 'main',url: 'https://github.com/your-org/your-python-project.git'}}stage('Setup Environment') {steps {sh '''python3.12 -m pip install --upgrade pippython3.12 -m pip install poetrypoetry install --no-root --with dev'''}}stage('Run Tests') {steps {sh '''poetry run pytest tests/ -v --junitxml=test-results.xml'''}post {always {junit 'test-results.xml'}}}stage('Run Python Script') {steps {sh '''poetry run python src/main.py'''}}stage('Upload to S3') {steps {sh '''aws s3 sync ./output/ s3://${S3_BUCKET}/${JOB_NAME}/${BUILD_NUMBER}/'''}}}post {always {publishHTML([allowMissing: false,alwaysLinkToLastBuild: false,keepAll: true,reportDir: 'reports',reportFiles: '*.html',reportName: 'Test Report'])}success {script {emailext (subject: "SUCCESS: Job '${env.JOB_NAME} [${env.BUILD_NUMBER}]'",body: """The build ${currentBuild.fullDisplayName} completed successfully!Check changeset at: ${env.BUILD_URL}/changesCheck console output at: ${env.BUILD_URL}/console""",to: "devops@yourcompany.com")}}failure {script {emailext (subject: "FAILED: Job '${env.JOB_NAME} [${env.BUILD_NUMBER}]'",body: """The build ${currentBuild.fullDisplayName} failed!Check changeset at: ${env.BUILD_URL}/changesCheck console output at: ${env.BUILD_URL}/console""",to: "devops@yourcompany.com")}}}
}