144 lines
5.7 KiB
Python
144 lines
5.7 KiB
Python
"""
|
|
클라우드 전환 자동화 체크리스트 — On-prem → K-Cloud/NCloud
|
|
|
|
공공기관 클라우드 전환 준비도 점검 및 체크리스트 자동 생성.
|
|
|
|
엔드포인트:
|
|
GET /api/migration/checklist — 전환 체크리스트 조회
|
|
POST /api/migration/assess — 전환 준비도 평가
|
|
GET /api/migration/readiness — 준비도 현황
|
|
PUT /api/migration/checklist/{key} — 체크리스트 항목 완료 처리
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from core.auth import get_current_user
|
|
from database import get_db
|
|
from models import User, MigrationChecklist
|
|
|
|
router = APIRouter(prefix="/api/migration", tags=["클라우드 전환"])
|
|
|
|
MIGRATION_PHASES = {
|
|
"01_사전평가": [
|
|
{"key": "infra_inventory", "title": "현행 인프라 인벤토리 완성", "priority": "MUST"},
|
|
{"key": "app_compatibility", "title": "애플리케이션 클라우드 호환성 평가", "priority": "MUST"},
|
|
{"key": "network_bandwidth", "title": "네트워크 대역폭 요구사항 측정", "priority": "MUST"},
|
|
{"key": "license_check", "title": "SW 라이선스 클라우드 이전 가능 여부 확인", "priority": "MUST"},
|
|
{"key": "cost_estimate", "title": "TCO(총소유비용) 분석 완료", "priority": "RECOMMEND"},
|
|
],
|
|
"02_보안준비": [
|
|
{"key": "iam_design", "title": "IAM 역할·정책 설계 완료", "priority": "MUST"},
|
|
{"key": "vpc_design", "title": "VPC·서브넷·보안그룹 설계", "priority": "MUST"},
|
|
{"key": "kms_setup", "title": "암호화 키 관리(KMS) 설정", "priority": "MUST"},
|
|
{"key": "csap_check", "title": "CSAP(클라우드 보안인증) 인증 CSP 확인", "priority": "MUST"},
|
|
{"key": "zero_trust", "title": "제로트러스트 보안 정책 수립", "priority": "RECOMMEND"},
|
|
],
|
|
"03_이전실행": [
|
|
{"key": "data_migration_plan", "title": "데이터 마이그레이션 계획 수립", "priority": "MUST"},
|
|
{"key": "blue_green", "title": "Blue/Green 전환 전략 수립", "priority": "RECOMMEND"},
|
|
{"key": "rollback_plan", "title": "롤백 계획 수립", "priority": "MUST"},
|
|
{"key": "downtime_plan", "title": "서비스 중단 최소화 방안 확보", "priority": "MUST"},
|
|
{"key": "dr_config", "title": "클라우드 DR 구성 완료", "priority": "RECOMMEND"},
|
|
],
|
|
"04_운영전환": [
|
|
{"key": "monitoring_setup", "title": "클라우드 모니터링 설정", "priority": "MUST"},
|
|
{"key": "staff_training", "title": "운영 인력 교육 완료", "priority": "MUST"},
|
|
{"key": "runbook_update", "title": "운영 매뉴얼 업데이트", "priority": "RECOMMEND"},
|
|
{"key": "cost_governance", "title": "비용 거버넌스 정책 수립", "priority": "RECOMMEND"},
|
|
],
|
|
}
|
|
|
|
|
|
class ChecklistUpdate(BaseModel):
|
|
completed: bool
|
|
note: Optional[str] = None
|
|
|
|
|
|
@router.get("/checklist")
|
|
async def get_checklist(
|
|
db: AsyncSession = Depends(get_db),
|
|
user: User = Depends(get_current_user),
|
|
):
|
|
rows = await db.execute(
|
|
select(MigrationChecklist).where(MigrationChecklist.tenant_id == user.tenant_id)
|
|
)
|
|
completed_keys = {c.item_key for c in rows.scalars().all() if c.completed}
|
|
|
|
result = {}
|
|
for phase, items in MIGRATION_PHASES.items():
|
|
result[phase] = [
|
|
{**item, "completed": item["key"] in completed_keys}
|
|
for item in items
|
|
]
|
|
|
|
total = sum(len(v) for v in MIGRATION_PHASES.values())
|
|
completed = len(completed_keys)
|
|
return {
|
|
"phases": result,
|
|
"progress": {"completed": completed, "total": total,
|
|
"pct": round(completed / total * 100, 1) if total else 0},
|
|
}
|
|
|
|
|
|
@router.put("/checklist/{item_key}")
|
|
async def update_checklist_item(
|
|
item_key: str,
|
|
req: ChecklistUpdate,
|
|
db: AsyncSession = Depends(get_db),
|
|
user: User = Depends(get_current_user),
|
|
):
|
|
row = await db.execute(
|
|
select(MigrationChecklist).where(
|
|
MigrationChecklist.tenant_id == user.tenant_id,
|
|
MigrationChecklist.item_key == item_key,
|
|
)
|
|
)
|
|
item = row.scalar_one_or_none()
|
|
if item:
|
|
item.completed = req.completed
|
|
item.note = req.note
|
|
item.updated_at = datetime.utcnow()
|
|
else:
|
|
item = MigrationChecklist(
|
|
tenant_id=user.tenant_id, item_key=item_key,
|
|
completed=req.completed, note=req.note,
|
|
created_at=datetime.utcnow(), updated_at=datetime.utcnow(),
|
|
)
|
|
db.add(item)
|
|
await db.commit()
|
|
return {"ok": True, "item_key": item_key, "completed": req.completed}
|
|
|
|
|
|
@router.get("/readiness")
|
|
async def migration_readiness(
|
|
db: AsyncSession = Depends(get_db),
|
|
user: User = Depends(get_current_user),
|
|
):
|
|
checklist = await get_checklist(db, user)
|
|
progress = checklist["progress"]
|
|
pct = progress["pct"]
|
|
phase_status = {}
|
|
for phase, items in checklist["phases"].items():
|
|
total = len(items)
|
|
done = sum(1 for i in items if i["completed"])
|
|
must_total = sum(1 for i in items if i["priority"] == "MUST")
|
|
must_done = sum(1 for i in items if i["priority"] == "MUST" and i["completed"])
|
|
phase_status[phase] = {
|
|
"progress": f"{done}/{total}",
|
|
"must_completed": must_done == must_total,
|
|
"status": "READY" if must_done == must_total else "IN_PROGRESS",
|
|
}
|
|
return {
|
|
"overall_pct": pct,
|
|
"readiness": "READY" if pct >= 80 else "IN_PROGRESS" if pct >= 40 else "NOT_STARTED",
|
|
"phases": phase_status,
|
|
}
|