// AWS Cognito User Poolsのカスタム属性で8種類のユーザー属性を管理
{
  "custom:user_type": "lumber_mill",  // 製材所
  "custom:permissions": "create_order,view_inventory",
  "custom:company_id": "company-123"
}

# Flask + AWS Cognito統合による認証・認可
from functools import wraps
from flask import request, jsonify

def require_user_type(*allowed_types):
    """ユーザー属性ごとのアクセス制御デコレータ"""
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            user_type = get_user_type_from_token(request.headers.get('Authorization'))

            if user_type not in allowed_types:
                return jsonify({'error': 'Forbidden'}), 403

            return f(*args, **kwargs)
        return decorated_function
    return decorator

@app.route('/api/inventory', methods=['GET'])
@require_user_type('lumber_mill', 'market', 'manufacturer')
def get_inventory():
    """在庫情報取得（製材所・市場・メーカーのみアクセス可）"""
    user_company_id = get_company_id_from_token()
    # 自社の在庫のみ取得（他社データは取得不可）
    inventory = Inventory.query.filter_by(company_id=user_company_id).all()
    return jsonify([inv.to_dict() for inv in inventory])

import { z } from "zod";

const OrderSchema = z.object({
  product_id: z.string().uuid(),
  quantity: z.number().int().positive().max(10000),
  unit_price: z.number().positive().max(1000000),
  delivery_date: z.string().datetime(),
});

type Order = z.infer<typeof OrderSchema>;

// フォーム送信前にバリデーション
const handleSubmit = (data: Order) => {
  const result = OrderSchema.safeParse(data);
  if (!result.success) {
    alert(result.error.errors[0].message);
    return;
  }
  // API送信
};

from marshmallow import Schema, fields, validate, ValidationError

class OrderSchema(Schema):
    product_id = fields.UUID(required=True)
    quantity = fields.Integer(required=True, validate=validate.Range(min=1, max=10000))
    unit_price = fields.Decimal(required=True, validate=validate.Range(min=0, max=1000000))
    delivery_date = fields.DateTime(required=True)

@app.route('/api/orders', methods=['POST'])
def create_order():
    schema = OrderSchema()
    try:
        data = schema.load(request.json)
    except ValidationError as err:
        return jsonify({'errors': err.messages}), 400

    # DBに保存
    order = Order(**data)
    db.session.add(order)
    db.session.commit()
    return jsonify(order.to_dict()), 201

# VPC設定
resource "aws_vpc" "main" {
  cidr_block           = "10.0.0.0/16"
  enable_dns_hostnames = true
  enable_dns_support   = true

  tags = {
    Name        = "${var.project_name}-vpc"
    Environment = var.environment
  }
}

# ECS on Fargate
resource "aws_ecs_cluster" "main" {
  name = "${var.project_name}-cluster"
}

resource "aws_ecs_service" "app" {
  name            = "${var.project_name}-app"
  cluster         = aws_ecs_cluster.main.id
  task_definition = aws_ecs_task_definition.app.arn
  desired_count   = var.environment == "production" ? 3 : 1
  launch_type     = "FARGATE"

  network_configuration {
    subnets         = aws_subnet.private[*].id
    security_groups = [aws_security_group.ecs_tasks.id]
  }

  load_balancer {
    target_group_arn = aws_lb_target_group.app.arn
    container_name   = "app"
    container_port   = 8080
  }
}

# AWS Cognito User Pool
resource "aws_cognito_user_pool" "main" {
  name = "${var.project_name}-user-pool"

  # 8種類のユーザー属性をカスタム属性で定義
  schema {
    name                = "user_type"
    attribute_data_type = "String"
    mutable             = true
  }

  schema {
    name                = "company_id"
    attribute_data_type = "String"
    mutable             = false
  }
}

from concurrent.futures import ThreadPoolExecutor, as_completed
from sqlalchemy.orm import selectinload

def parallel_create_documents(app, order_id: str) -> None:
    """注文書・納品書・請求書を同時生成する。"""
    tasks = [create_order_form, create_delivery_note, create_invoice]

    def run(task):
        with app.app_context():                       # スレッドごとにコンテキストを張る
            doc = (
                Document.query
                .options(selectinload(Document.lines))  # N+1 を選択ロードで回避
                .filter_by(order_id=order_id)
                .with_for_update()                       # 行ロックで競合生成を防ぐ
                .one()
            )
            return task(doc)                             # openpyxl で Excel → LibreOffice で PDF

    with ThreadPoolExecutor(max_workers=len(tasks)) as pool:
        futures = [pool.submit(run, t) for t in tasks]
        for f in as_completed(futures):
            f.result()  # 最初の例外を伝播させる

import stripe, hashlib

stripe.api_key = os.getenv("STRIPE_SECRET_KEY")

def create_subscription(customer: User, plan_id: str) -> dict:
    # 金額は plan_id からサーバ側で解決する（クライアント指定額を信用しない＝改ざん対策）
    subscription = stripe.Subscription.create(
        customer=customer.stripe_customer_id,
        items=[{"price": plan_id}],
        payment_behavior="default_incomplete",
        expand=["latest_invoice.payment_intent"],
        # 同一内容の再送が二重課金にならないよう冪等キーを付ける
        idempotency_key=f"sub_{customer.id}_{plan_id}",
    )
    return {"subscription_id": subscription.id}

# .github/workflows/deploy.yml
name: Deploy to ECS

on:
  push:
    branches: [main]

jobs:
  test-and-deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      # リンター実行
      - name: Run ESLint
        run: npm run lint

      - name: Run Flake8
        run: flake8 app/

      # 脆弱性診断
      - name: Security Audit
        run: |
          npm audit --production
          pip-audit

      # テスト実行
      - name: Run Tests
        run: pytest tests/

      # Dockerイメージビルド
      - name: Build Docker Image
        run: docker build -t my-app:${{ github.sha }} .

      # ECRにプッシュ
      - name: Push to ECR
        run: |
          aws ecr get-login-password --region ap-northeast-1 | docker login --username AWS --password-stdin ${{ secrets.ECR_REGISTRY }}
          docker push my-app:${{ github.sha }}

      # ECS デプロイ
      - name: Deploy to ECS
        run: |
          aws ecs update-service --cluster my-cluster --service my-app --force-new-deployment