# 特性1: オンデマンド・セルフサービス — boto3でEC2インスタンスを即時起動
import boto3
 
ec2 = boto3.resource('ec2', region_name='ap-northeast-1')
 
# 開発者が自分のタイミングでリソースを確保
instances = ec2.create_instances(
    ImageId='ami-0abcdef1234567890',
    MinCount=1,
    MaxCount=1,
    InstanceType='t3.micro',
    TagSpecifications=[{
        'ResourceType': 'instance',
        'Tags': [{'Key': 'Name', 'Value': 'dev-server'}]
    }]
)
print(f"インスタンス {instances[0].id} を起動しました")

# 特性4: 迅速な弾力性 — Auto Scalingグループの設定
import boto3
 
autoscaling = boto3.client('autoscaling', region_name='ap-northeast-1')
 
# 需要に応じて1〜10台の範囲で自動スケール
autoscaling.create_auto_scaling_group(
    AutoScalingGroupName='web-asg',
    LaunchTemplate={
        'LaunchTemplateId': 'lt-0abcdef1234567890',
        'Version': '$Latest'
    },
    MinSize=1,
    MaxSize=10,
    DesiredCapacity=2,
    VPCZoneIdentifier='subnet-abc123,subnet-def456',
    TargetGroupARNs=['arn:aws:elasticloadbalancing:...'],
    Tags=[{
        'Key': 'Environment',
        'Value': 'production',
        'PropagateAtLaunch': True
    }]
)
 
# スケーリングポリシーの設定（CPU使用率70%を基準）
autoscaling.put_scaling_policy(
    AutoScalingGroupName='web-asg',
    PolicyName='cpu-target-tracking',
    PolicyType='TargetTrackingScaling',
    TargetTrackingConfiguration={
        'PredefinedMetricSpecification': {
            'PredefinedMetricType': 'ASGAverageCPUUtilization'
        },
        'TargetValue': 70.0,
        'ScaleInCooldown': 300,
        'ScaleOutCooldown': 60
    }
)

# 特性5: 従量課金 — コスト監視と予算アラート設定
import boto3
 
budgets = boto3.client('budgets', region_name='us-east-1')
 
# 月額予算を設定し、閾値超過時に通知
budgets.create_budget(
    AccountId='123456789012',
    Budget={
        'BudgetName': 'monthly-budget',
        'BudgetLimit': {
            'Amount': '1000',
            'Unit': 'USD'
        },
        'BudgetType': 'COST',
        'TimeUnit': 'MONTHLY',
        'TimePeriod': {
            'Start': '2026-01-01T00:00:00Z',
            'End': '2027-01-01T00:00:00Z'
        }
    },
    NotificationsWithSubscribers=[{
        'Notification': {
            'NotificationType': 'ACTUAL',
            'ComparisonOperator': 'GREATER_THAN',
            'Threshold': 80.0,
            'ThresholdType': 'PERCENTAGE'
        },
        'Subscribers': [{
            'SubscriptionType': 'EMAIL',
            'Address': 'admin@example.com'
        }]
    }]
)

# TCO計算スクリプト
def calculate_tco_comparison(years: int = 3):
    """オンプレミスとクラウドのTCO比較"""
 
    # オンプレミスコスト
    onprem = {
        'hardware': {
            'servers': 2_000_000 * 10,
            'network': 5_000_000,
            'storage': 3_000_000,
        },
        'monthly': {
            'datacenter': 500_000,
            'power_cooling': 200_000,
            'staff': 1_200_000,
            'oncall': 300_000,
        },
        'yearly': {
            'licenses': 2_000_000,
            'maintenance': 1_500_000,  # ハードウェア保守
        }
    }
 
    onprem_total = (
        sum(onprem['hardware'].values()) +
        sum(onprem['monthly'].values()) * 12 * years +
        sum(onprem['yearly'].values()) * years
    )
 
    # クラウドコスト
    cloud = {
        'monthly': {
            'compute': 300_000,
            'serverless': 50_000,
            'storage_db': 150_000,
            'cache': 80_000,
            'network': 100_000,
            'staff': 700_000,
        },
        'yearly': {
            'support_plan': 600_000,
            'training': 300_000,
        }
    }
 
    cloud_total = (
        sum(cloud['monthly'].values()) * 12 * years +
        sum(cloud['yearly'].values()) * years
    )
 
    savings = onprem_total - cloud_total
    savings_pct = (savings / onprem_total) * 100
 
    print(f"=== {years}年間TCO比較 ===")
    print(f"オンプレミス: ¥{onprem_total:,.0f}")
    print(f"クラウド:     ¥{cloud_total:,.0f}")
    print(f"削減額:       ¥{savings:,.0f} ({savings_pct:.1f}%)")
 
    return {
        'onprem': onprem_total,
        'cloud': cloud_total,
        'savings': savings,
        'savings_pct': savings_pct
    }
 
result = calculate_tco_comparison(3)
# === 3年間TCO比較 ===
# オンプレミス: ¥117,400,000
# クラウド:     ¥52,680,000
# 削減額:       ¥64,720,000 (55.1%)

# EC2インスタンスを起動する最小コマンド
aws ec2 run-instances \
  --image-id ami-0abcdef1234567890 \
  --instance-type t3.micro \
  --key-name my-key-pair \
  --security-group-ids sg-0123456789abcdef0 \
  --subnet-id subnet-0123456789abcdef0 \
  --count 1 \
  --tag-specifications 'ResourceType=instance,Tags=[{Key=Name,Value=web-server-01}]'
 
# インスタンスの状態確認
aws ec2 describe-instances \
  --filters "Name=tag:Name,Values=web-server-01" \
  --query 'Reservations[].Instances[].[InstanceId,State.Name,PublicIpAddress]' \
  --output table

# Terraform (IaC) でEC2を定義する例
# main.tf
"""
resource "aws_instance" "web_server" {
  ami           = "ami-0abcdef1234567890"
  instance_type = "t3.micro"
  subnet_id     = aws_subnet.public.id
 
  vpc_security_group_ids = [aws_security_group.web.id]
 
  user_data = <<-EOF
    #!/bin/bash
    yum update -y
    yum install -y httpd
    systemctl start httpd
    systemctl enable httpd
    echo "<h1>Hello from EC2</h1>" > /var/www/html/index.html
  EOF
 
  tags = {
    Name        = "web-server-01"
    Environment = "production"
    ManagedBy   = "terraform"
  }
 
  root_block_device {
    volume_size = 20
    volume_type = "gp3"
    encrypted   = true
  }
}
"""

// ECSタスク定義 (task-definition.json)
{
  "family": "web-app",
  "networkMode": "awsvpc",
  "requiresCompatibilities": ["FARGATE"],
  "cpu": "256",
  "memory": "512",
  "executionRoleArn": "arn:aws:iam::123456789012:role/ecsTaskExecutionRole",
  "containerDefinitions": [
    {
      "name": "web",
      "image": "123456789012.dkr.ecr.ap-northeast-1.amazonaws.com/web-app:latest",
      "portMappings": [
        {
          "containerPort": 8080,
          "protocol": "tcp"
        }
      ],
      "logConfiguration": {
        "logDriver": "awslogs",
        "options": {
          "awslogs-group": "/ecs/web-app",
          "awslogs-region": "ap-northeast-1",
          "awslogs-stream-prefix": "ecs"
        }
      },
      "environment": [
        {"name": "NODE_ENV", "value": "production"},
        {"name": "PORT", "value": "8080"}
      ],
      "healthCheck": {
        "command": ["CMD-SHELL", "curl -f http://localhost:8080/health || exit 1"],
        "interval": 30,
        "timeout": 5,
        "retries": 3,
        "startPeriod": 60
      }
    }
  ]
}

# ECSサービスをFargateで作成
aws ecs create-service \
  --cluster production \
  --service-name web-app \
  --task-definition web-app:1 \
  --desired-count 2 \
  --launch-type FARGATE \
  --network-configuration "awsvpcConfiguration={
    subnets=[subnet-abc123,subnet-def456],
    securityGroups=[sg-web123],
    assignPublicIp=DISABLED
  }" \
  --load-balancers "targetGroupArn=arn:aws:elasticloadbalancing:...,containerName=web,containerPort=8080"

# Elastic Beanstalk CLIで環境を作成
eb init my-app --platform python-3.11 --region ap-northeast-1
eb create my-env --instance-type t3.small --envvars DATABASE_URL=postgresql://...
eb deploy
 
# 環境変数の更新
eb setenv SECRET_KEY=my-secret-key DEBUG=false
 
# ログの確認
eb logs --all

# .ebextensions/01-packages.config
packages:
  yum:
    postgresql-devel: []
 
container_commands:
  01_migrate:
    command: "python manage.py migrate"
    leader_only: true
  02_collectstatic:
    command: "python manage.py collectstatic --noinput"
 
option_settings:
  aws:elasticbeanstalk:container:python:
    WSGIPath: myapp.wsgi:application
  aws:elasticbeanstalk:environment:proxy:staticfiles:
    /static: static
  aws:autoscaling:asg:
    MinSize: 2
    MaxSize: 8
  aws:autoscaling:trigger:
    MeasureName: CPUUtilization
    Statistic: Average
    Unit: Percent
    UpperThreshold: 70
    LowerThreshold: 30

# Lambda関数: S3にアップロードされた画像をリサイズ
import json
import boto3
from PIL import Image
import io
import os
 
s3 = boto3.client('s3')
 
def lambda_handler(event, context):
    """S3イベントトリガーで画像をリサイズする"""
 
    # イベントからバケット名とキーを取得
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key']
 
    # 既にリサイズ済みの画像はスキップ
    if key.startswith('thumbnails/'):
        return {'statusCode': 200, 'body': 'Already processed'}
 
    try:
        # 元画像をダウンロード
        response = s3.get_object(Bucket=bucket, Key=key)
        image_content = response['Body'].read()
 
        # リサイズ処理
        image = Image.open(io.BytesIO(image_content))
        image.thumbnail((300, 300), Image.LANCZOS)
 
        # リサイズした画像をバッファに保存
        buffer = io.BytesIO()
        image.save(buffer, format='JPEG', quality=85)
        buffer.seek(0)
 
        # サムネイルをS3にアップロード
        thumbnail_key = f"thumbnails/{os.path.basename(key)}"
        s3.put_object(
            Bucket=bucket,
            Key=thumbnail_key,
            Body=buffer,
            ContentType='image/jpeg',
            CacheControl='max-age=31536000'
        )
 
        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Thumbnail created',
                'source': key,
                'thumbnail': thumbnail_key,
                'original_size': f"{image.size[0]}x{image.size[1]}"
            })
        }
 
    except Exception as e:
        print(f"Error processing {key}: {str(e)}")
        raise

# AWS SAM テンプレート (template.yaml)
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: Image Resize Lambda Function
 
Globals:
  Function:
    Timeout: 30
    Runtime: python3.11
    MemorySize: 512
    Architectures:
      - arm64
 
Resources:
  ImageBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub "${AWS::StackName}-images"
 
  ResizeFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: resize.lambda_handler
      CodeUri: src/
      Policies:
        - S3CrudPolicy:
            BucketName: !Ref ImageBucket
      Events:
        S3Upload:
          Type: S3
          Properties:
            Bucket: !Ref ImageBucket
            Events: s3:ObjectCreated:*
            Filter:
              S3Key:
                Rules:
                  - Name: prefix
                    Value: uploads/
                  - Name: suffix
                    Value: .jpg

import boto3
from botocore.exceptions import ClientError
 
def send_templated_email(
    sender: str,
    recipient: str,
    template_name: str,
    template_data: dict
) -> dict:
    """SESテンプレートメールを送信する"""
 
    client = boto3.client('ses', region_name='ap-northeast-1')
 
    try:
        response = client.send_templated_email(
            Source=sender,
            Destination={
                'ToAddresses': [recipient],
            },
            Template=template_name,
            TemplateData=json.dumps(template_data),
            ConfigurationSetName='tracking-config',
            Tags=[
                {'Name': 'campaign', 'Value': 'welcome'},
                {'Name': 'environment', 'Value': 'production'},
            ]
        )
        print(f"Email sent! Message ID: {response['MessageId']}")
        return response
 
    except ClientError as e:
        error_code = e.response['Error']['Code']
        if error_code == 'MessageRejected':
            print(f"メール拒否: {e.response['Error']['Message']}")
        elif error_code == 'MailFromDomainNotVerifiedException':
            print("送信元ドメインが未検証です")
        else:
            print(f"予期しないエラー: {error_code}")
        raise
 
# 使用例
import json
send_templated_email(
    sender='noreply@example.com',
    recipient='user@example.com',
    template_name='WelcomeEmail',
    template_data={
        'name': '田中太郎',
        'company': 'Example Corp',
        'login_url': 'https://app.example.com/login'
    }
)

# AWS Direct Connect + VPNの構成例
# Direct Connect Gateway の作成
aws directconnect create-direct-connect-gateway \
  --direct-connect-gateway-name "onprem-gateway" \
  --amazon-side-asn 64512
 
# VPN接続の作成（バックアップ用）
aws ec2 create-vpn-gateway \
  --type ipsec.1 \
  --amazon-side-asn 65000
 
aws ec2 create-customer-gateway \
  --type ipsec.1 \
  --public-ip 203.0.113.1 \
  --bgp-asn 65100
 
aws ec2 create-vpn-connection \
  --type ipsec.1 \
  --vpn-gateway-id vgw-abc123 \
  --customer-gateway-id cgw-def456 \
  --options '{"StaticRoutesOnly": false}'

# マルチクラウド抽象化レイヤーの例
from abc import ABC, abstractmethod
from typing import BinaryIO
 
class CloudStorageProvider(ABC):
    """クラウドストレージの抽象基底クラス"""
 
    @abstractmethod
    def upload(self, bucket: str, key: str, data: BinaryIO) -> str:
        pass
 
    @abstractmethod
    def download(self, bucket: str, key: str) -> bytes:
        pass
 
    @abstractmethod
    def delete(self, bucket: str, key: str) -> bool:
        pass
 
    @abstractmethod
    def list_objects(self, bucket: str, prefix: str = '') -> list:
        pass
 
class AWSS3Provider(CloudStorageProvider):
    def __init__(self, region: str = 'ap-northeast-1'):
        import boto3
        self.client = boto3.client('s3', region_name=region)
 
    def upload(self, bucket: str, key: str, data: BinaryIO) -> str:
        self.client.upload_fileobj(data, bucket, key)
        return f"s3://{bucket}/{key}"
 
    def download(self, bucket: str, key: str) -> bytes:
        response = self.client.get_object(Bucket=bucket, Key=key)
        return response['Body'].read()
 
    def delete(self, bucket: str, key: str) -> bool:
        self.client.delete_object(Bucket=bucket, Key=key)
        return True
 
    def list_objects(self, bucket: str, prefix: str = '') -> list:
        response = self.client.list_objects_v2(Bucket=bucket, Prefix=prefix)
        return [obj['Key'] for obj in response.get('Contents', [])]
 
class GCPStorageProvider(CloudStorageProvider):
    def __init__(self):
        from google.cloud import storage
        self.client = storage.Client()
 
    def upload(self, bucket: str, key: str, data: BinaryIO) -> str:
        bucket_obj = self.client.bucket(bucket)
        blob = bucket_obj.blob(key)
        blob.upload_from_file(data)
        return f"gs://{bucket}/{key}"
 
    def download(self, bucket: str, key: str) -> bytes:
        bucket_obj = self.client.bucket(bucket)
        blob = bucket_obj.blob(key)
        return blob.download_as_bytes()
 
    def delete(self, bucket: str, key: str) -> bool:
        bucket_obj = self.client.bucket(bucket)
        blob = bucket_obj.blob(key)
        blob.delete()
        return True
 
    def list_objects(self, bucket: str, prefix: str = '') -> list:
        bucket_obj = self.client.bucket(bucket)
        blobs = bucket_obj.list_blobs(prefix=prefix)
        return [blob.name for blob in blobs]
 
class AzureBlobProvider(CloudStorageProvider):
    def __init__(self, connection_string: str):
        from azure.storage.blob import BlobServiceClient
        self.client = BlobServiceClient.from_connection_string(connection_string)
 
    def upload(self, bucket: str, key: str, data: BinaryIO) -> str:
        blob_client = self.client.get_blob_client(bucket, key)
        blob_client.upload_blob(data, overwrite=True)
        return f"https://{self.client.account_name}.blob.core.windows.net/{bucket}/{key}"
 
    def download(self, bucket: str, key: str) -> bytes:
        blob_client = self.client.get_blob_client(bucket, key)
        return blob_client.download_blob().readall()
 
    def delete(self, bucket: str, key: str) -> bool:
        blob_client = self.client.get_blob_client(bucket, key)
        blob_client.delete_blob()
        return True
 
    def list_objects(self, bucket: str, prefix: str = '') -> list:
        container = self.client.get_container_client(bucket)
        blobs = container.list_blobs(name_starts_with=prefix)
        return [blob.name for blob in blobs]
 
# ファクトリパターンでプロバイダを選択
def get_storage_provider(cloud: str, **kwargs) -> CloudStorageProvider:
    providers = {
        'aws': AWSS3Provider,
        'gcp': GCPStorageProvider,
        'azure': AzureBlobProvider,
    }
    if cloud not in providers:
        raise ValueError(f"Unsupported cloud: {cloud}")
    return providerscloud
 
# 使用例
provider = get_storage_provider('aws', region='ap-northeast-1')
url = provider.upload('my-bucket', 'data/report.csv', open('report.csv', 'rb'))

# AWS
aws s3 mb s3://my-bucket-2026 --region ap-northeast-1
aws s3api put-bucket-versioning \
  --bucket my-bucket-2026 \
  --versioning-configuration Status=Enabled
aws s3api put-bucket-encryption \
  --bucket my-bucket-2026 \
  --server-side-encryption-configuration '{
    "Rules": [{"ApplyServerSideEncryptionByDefault": {"SSEAlgorithm": "aws:kms"}}]
  }'
 
# GCP
gsutil mb -l asia-northeast1 gs://my-bucket-2026
gsutil versioning set on gs://my-bucket-2026
 
# Azure
az storage account create \
  --name mystorageaccount \
  --resource-group myResourceGroup \
  --location japaneast \
  --sku Standard_LRS
az storage container create \
  --name my-bucket-2026 \
  --account-name mystorageaccount

# === AWS S3 ===
import boto3
 
s3 = boto3.client('s3')
 
# アップロード（メタデータ付き）
s3.upload_file(
    'local.txt',
    'my-bucket',
    'remote.txt',
    ExtraArgs={
        'ContentType': 'text/plain',
        'ServerSideEncryption': 'aws:kms',
        'Metadata': {
            'uploaded-by': 'automation',
            'version': '1.0'
        }
    }
)
 
# プリサインドURLの生成（一時的な共有用）
presigned_url = s3.generate_presigned_url(
    'get_object',
    Params={'Bucket': 'my-bucket', 'Key': 'remote.txt'},
    ExpiresIn=3600  # 1時間有効
)
print(f"共有URL: {presigned_url}")
 
# === GCP Cloud Storage ===
from google.cloud import storage
 
client = storage.Client()
bucket = client.bucket('my-bucket')
blob = bucket.blob('remote.txt')
blob.metadata = {'uploaded-by': 'automation', 'version': '1.0'}
blob.upload_from_filename('local.txt', content_type='text/plain')
 
# 署名付きURLの生成
signed_url = blob.generate_signed_url(
    version='v4',
    expiration=3600,
    method='GET'
)
 
# === Azure Blob Storage ===
from azure.storage.blob import BlobServiceClient, generate_blob_sas, BlobSasPermissions
from datetime import datetime, timedelta
 
blob_service = BlobServiceClient.from_connection_string(conn_str)
blob_client = blob_service.get_blob_client('my-container', 'remote.txt')
 
with open('local.txt', 'rb') as data:
    blob_client.upload_blob(
        data,
        content_settings={'content_type': 'text/plain'},
        metadata={'uploaded-by': 'automation', 'version': '1.0'}
    )
 
# SAS URLの生成
sas_token = generate_blob_sas(
    account_name='mystorageaccount',
    container_name='my-container',
    blob_name='remote.txt',
    account_key='...',
    permission=BlobSasPermissions(read=True),
    expiry=datetime.utcnow() + timedelta(hours=1)
)

# 12 Factor App 準拠のアプリケーション例
 
# Factor 3: 設定は環境変数から
import os
 
class Config:
    DATABASE_URL = os.environ['DATABASE_URL']        # 必須
    REDIS_URL = os.environ.get('REDIS_URL', 'redis://localhost:6379')
    SECRET_KEY = os.environ['SECRET_KEY']
    DEBUG = os.environ.get('DEBUG', 'false').lower() == 'true'
    LOG_LEVEL = os.environ.get('LOG_LEVEL', 'INFO')
    PORT = int(os.environ.get('PORT', '8080'))
 
# Factor 4: バッキングサービスをアタッチされたリソースとして
# → DB接続先は環境変数で注入、コード変更なしで切替可能
 
# Factor 6: ステートレスプロセス
# → セッション情報はRedisに保持、ローカルファイルに書かない
 
# Factor 7: ポートバインディング
from fastapi import FastAPI
import uvicorn
 
app = FastAPI()
 
@app.get("/health")
def health():
    return {"status": "healthy"}
 
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=Config.PORT)
 
# Factor 9: 廃棄容易性 — グレースフルシャットダウン
import signal
import sys
 
def graceful_shutdown(signum, frame):
    print("Shutting down gracefully...")
    # DB接続のクローズ、進行中のリクエスト完了を待つ
    sys.exit(0)
 
signal.signal(signal.SIGTERM, graceful_shutdown)
signal.signal(signal.SIGINT, graceful_shutdown)
 
# Factor 11: ログはstdoutに出力
import logging
import json
 
class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            'timestamp': self.formatTime(record),
            'level': record.levelname,
            'message': record.getMessage(),
            'logger': record.name,
            'module': record.module,
        }
        if record.exc_info:
            log_data['exception'] = self.formatException(record.exc_info)
        return json.dumps(log_data, ensure_ascii=False)
 
handler = logging.StreamHandler(sys.stdout)  # stdoutに出力
handler.setFormatter(JSONFormatter())
logger = logging.getLogger(__name__)
logger.addHandler(handler)
logger.setLevel(Config.LOG_LEVEL)

# AWSコスト最適化ツール: Cost Explorer APIの活用
import boto3
from datetime import datetime, timedelta
 
ce = boto3.client('ce', region_name='us-east-1')
 
def get_monthly_cost_breakdown():
    """月別・サービス別コスト内訳を取得"""
    end = datetime.now().strftime('%Y-%m-%d')
    start = (datetime.now() - timedelta(days=90)).strftime('%Y-%m-%d')
 
    response = ce.get_cost_and_usage(
        TimePeriod={'Start': start, 'End': end},
        Granularity='MONTHLY',
        Metrics=['UnblendedCost'],
        GroupBy=[
            {'Type': 'DIMENSION', 'Key': 'SERVICE'}
        ]
    )
 
    for period in response['ResultsByTime']:
        print(f"\n=== {period['TimePeriod']['Start']} ===")
        groups = sorted(
            period['Groups'],
            key=lambda x: float(x['Metrics']['UnblendedCost']['Amount']),
            reverse=True
        )
        for group in groups[:10]:
            service = group['Keys'][0]
            cost = float(group['Metrics']['UnblendedCost']['Amount'])
            if cost > 0:
                print(f"  {service}: ${cost:.2f}")
 
def get_savings_recommendations():
    """Savings Plans / Reserved Instances の推奨を取得"""
    response = ce.get_savings_plans_purchase_recommendation(
        SavingsPlansType='COMPUTE_SP',
        TermInYears='ONE_YEAR',
        PaymentOption='NO_UPFRONT',
        LookbackPeriodInDays='SIXTY_DAYS'
    )
 
    recommendation = response['SavingsPlansRecommendationDetails']
    for rec in recommendation[:5]:
        print(f"推奨月額: ${float(rec['HourlyCommitmentToPurchase']) * 730:.2f}")
        print(f"推定節約額: ${float(rec['EstimatedMonthlySavingsAmount']):.2f}")
        print(f"節約率: {float(rec['EstimatedSavingsPercentage']):.1f}%")
        print("---")
 
get_monthly_cost_breakdown()
get_savings_recommendations()

# AWS Migration Hub を使った移行進捗管理
import boto3
 
mh = boto3.client('migration-hub', region_name='us-west-2')
 
# 移行タスクの作成
def create_migration_task(app_name: str, strategy: str):
    """移行タスクを作成してトラッキングする"""
 
    mh.notify_migration_task_state(
        ProgressUpdateStream='MyMigrationStream',
        MigrationTaskName=f'{app_name}-migration',
        Task={
            'Status': 'IN_PROGRESS',
            'StatusDetail': f'Strategy: {strategy}',
            'ProgressPercent': 0
        },
        UpdateDateTime=datetime.now(),
        NextUpdateSeconds=3600
    )
 
    return f'{app_name}-migration'
 
# 移行進捗の更新
def update_migration_progress(task_name: str, percent: int, detail: str):
    mh.notify_migration_task_state(
        ProgressUpdateStream='MyMigrationStream',
        MigrationTaskName=task_name,
        Task={
            'Status': 'IN_PROGRESS' if percent < 100 else 'COMPLETED',
            'StatusDetail': detail,
            'ProgressPercent': percent
        },
        UpdateDateTime=datetime.now()
    )
 
# 使用例
task = create_migration_task('legacy-crm', 'Replatform')
update_migration_progress(task, 25, 'データベース移行中')
update_migration_progress(task, 50, 'アプリケーションデプロイ中')
update_migration_progress(task, 75, 'テスト実施中')
update_migration_progress(task, 100, '移行完了・本番稼働確認済み')

# 良い例: 初期段階からObservabilityを組み込む
import boto3
 
# CloudWatch アラームの設定
cloudwatch = boto3.client('cloudwatch', region_name='ap-northeast-1')
 
def setup_essential_alarms(instance_id: str):
    """EC2インスタンスの基本アラームをセットアップ"""
 
    alarms = [
        {
            'AlarmName': f'{instance_id}-high-cpu',
            'MetricName': 'CPUUtilization',
            'Namespace': 'AWS/EC2',
            'Statistic': 'Average',
            'Period': 300,
            'EvaluationPeriods': 3,
            'Threshold': 80.0,
            'ComparisonOperator': 'GreaterThanThreshold',
        },
        {
            'AlarmName': f'{instance_id}-status-check',
            'MetricName': 'StatusCheckFailed',
            'Namespace': 'AWS/EC2',
            'Statistic': 'Maximum',
            'Period': 60,
            'EvaluationPeriods': 2,
            'Threshold': 0,
            'ComparisonOperator': 'GreaterThanThreshold',
        },
    ]
 
    for alarm_config in alarms:
        cloudwatch.put_metric_alarm(
            **alarm_config,
            Dimensions=[{'Name': 'InstanceId', 'Value': instance_id}],
            AlarmActions=['arn:aws:sns:ap-northeast-1:123456789012:alerts'],
            TreatMissingData='breaching'
        )
        print(f"アラーム作成: {alarm_config['AlarmName']}")
 
setup_essential_alarms('i-0abcdef1234567890')

# 良い例: 体系的なタグ付け戦略
REQUIRED_TAGS = {
    'Environment': ['production', 'staging', 'development'],
    'Project': None,  # 自由入力
    'Owner': None,    # メールアドレス
    'CostCenter': None,
    'ManagedBy': ['terraform', 'cloudformation', 'manual'],
}
 
def validate_tags(tags: dict) -> list:
    """タグポリシーのバリデーション"""
    errors = []
    for required_key, allowed_values in REQUIRED_TAGS.items():
        if required_key not in tags:
            errors.append(f"必須タグ '{required_key}' がありません")
        elif allowed_values and tags[required_key] not in allowed_values:
            errors.append(
                f"タグ '{required_key}' の値 '{tags[required_key]}' は "
                f"許可値 {allowed_values} に含まれていません"
            )
    return errors
 
# AWS Config Rule でタグポリシーを強制
# required-tags ルールの設定
config_rule = {
    'ConfigRuleName': 'required-tags',
    'Source': {
        'Owner': 'AWS',
        'SourceIdentifier': 'REQUIRED_TAGS'
    },
    'InputParameters': json.dumps({
        'tag1Key': 'Environment',
        'tag2Key': 'Project',
        'tag3Key': 'Owner',
        'tag4Key': 'CostCenter'
    }),
    'Scope': {
        'ComplianceResourceTypes': [
            'AWS::EC2::Instance',
            'AWS::S3::Bucket',
            'AWS::RDS::DBInstance'
        ]
    }
}

# 共有責任モデルに基づくセキュリティチェックリスト
security_checklist = {
    'ユーザー責任': {
        'IAM': [
            'ルートアカウントにMFAを有効化',
            '個別IAMユーザーを作成（ルート共有禁止）',
            '最小権限の原則を適用',
            'IAMロールを使用（アクセスキー最小化）',
            'パスワードポリシーの強化',
            'アクセスキーの定期ローテーション',
        ],
        'データ保護': [
            'S3バケットのパブリックアクセスブロック',
            'EBSボリュームの暗号化',
            'RDSインスタンスの暗号化',
            'SSL/TLSの強制',
            'KMSによるキー管理',
        ],
        'ネットワーク': [
            'セキュリティグループの最小権限設定',
            'NACLの適切な設定',
            'VPCフローログの有効化',
            'プライベートサブネットの活用',
            'VPCエンドポイントの使用',
        ],
        '監視': [
            'CloudTrailの有効化（全リージョン）',
            'GuardDutyの有効化',
            'AWS Configの有効化',
            'CloudWatchアラームの設定',
            'Security Hubの統合',
        ]
    },
    'AWS責任': [
        '物理的データセンターセキュリティ',
        'ハードウェアの廃棄手順',
        'ネットワークインフラの保護',
        'ハイパーバイザーのセキュリティ',
        '電力・冷却の確保',
        'コンプライアンス認証の維持',
    ]
}
 
def print_checklist(checklist: dict, indent: int = 0):
    prefix = "  " * indent
    for key, value in checklist.items():
        print(f"{prefix}[ ] {key}")
        if isinstance(value, dict):
            print_checklist(value, indent + 1)
        elif isinstance(value, list):
            for item in value:
                print(f"{prefix}  [ ] {item}")
 
print_checklist(security_checklist)