feat: cli 세팅 및 깃허브 로그인 테스트

.pre-commit-config.yaml

@@ -9,4 +9,5 @@ repos:
   - repo: https://github.com/pre-commit/mirrors-mypy
     rev: v1.8.0
     hooks:
-      - id: mypy
+      - id: mypy
+        additional_dependencies: [types-requests, types-cachetools]

Makefile

@@ -3,7 +3,7 @@ FUNCTION_NAME ?= my-function-name
 .PHONY: dev build deploy
 
 dev:
-	uvicorn app.main:app --reload
+	uvicorn backend.main:app --reload
 
 build:
 	rm -rf ./_package lambda.zip

backend/adapters/inbound/api.py

@@ -1,12 +1,13 @@
-import os
+from typing import cast
 
+import requests
+from cachetools import TTLCache
 from fastapi import APIRouter, Depends, Header, HTTPException
 from pydantic import BaseModel
 
-from backend.adapters.outbound.dynamodb_repository import DynamoDBRepository
 from backend.application.exceptions import MemoAccessDeniedError
 from backend.application.ports import MemoUseCase
-from backend.application.service import MemoService
+from backend.dependencies import get_memo_use_case
 from backend.domain.memo import MemoDomainError
 
 router = APIRouter()
@@ -26,24 +27,56 @@ class UpdateMemoRequest(BaseModel):
     content: str
 
 
-def get_memo_use_case() -> MemoUseCase:
-    table_name = os.getenv("DYNAMODB_TABLE")
-    if not table_name:
-        raise ValueError("DYNAMODB_TABLE 환경변수가 설정되지 않았습니다")
+TOKEN_CACHE = cast(TTLCache[str, str, float], TTLCache(maxsize=100, ttl=300))
 
-    repo = DynamoDBRepository(table_name=table_name)
-    return MemoService(repo)
+
+def get_current_user_id(authorization: str | None = Header(None)) -> str:
+    if not authorization or not authorization.startswith("Bearer "):
+        raise HTTPException(
+            status_code=401,
+            detail="인증되지 않은 요청입니다. 유효한 Bearer 인증 토큰이 필요합니다.",
+        )
+
+    token = authorization.split(" ")[1]
+
+    if token in TOKEN_CACHE:
+        return TOKEN_CACHE[token]
+
+    github_user_uri = "https://api.github.com/user"
+    headers = {
+        "Authorization": f"Bearer {token}",
+        "Accept": "application/vnd.github+json",
+    }
+
+    try:
+        response = requests.get(github_user_uri, headers=headers, timeout=5)
+
+        if response.status_code != 200:
+            raise HTTPException(
+                status_code=401, detail="유효하지 않거나 만료된 깃허브 토큰입니다."
+            )
+
+        user_data = response.json()
+        user_id = str(user_data.get("id"))
+
+        TOKEN_CACHE[token] = user_id
+        return user_id
+
+    except requests.exceptions.RequestException:
+        raise HTTPException(
+            status_code=503, detail="깃허브 인증 서버와 통신할 수 없습니다."
+        )
 
 
-@router.post("/add", response_model=MemoCreateResponse)
+@router.post("/memos", response_model=MemoCreateResponse)
 def add_memo(
     request: MemoCreateRequest,
-    x_user_id: str = Header(..., alias="X-USER-ID"),
+    user_id: str = Depends(get_current_user_id),
     use_case: MemoUseCase = Depends(get_memo_use_case),
 ):
     try:
         memo = use_case.create_memo(
-            user_id=x_user_id,
+            user_id=user_id,
             content=request.content,
             category=request.category,
         )
@@ -56,11 +89,11 @@ def add_memo(
 def get_memos(
     category: str | None = None,
     limit: int = 5,
-    x_user_id: str = Header(..., alias="X-USER-ID"),
+    user_id: str = Depends(get_current_user_id),
     service: MemoUseCase = Depends(get_memo_use_case),
 ):
     memos = service.get_all_memos(
-        user_id=x_user_id,
+        user_id=user_id,
         category=category,
         limit=limit,
     )
@@ -70,12 +103,12 @@ def get_memos(
 @router.get("/memos/{memo_id}")
 def get_memo(
     memo_id: str,
-    x_user_id: str = Header(..., alias="X-USER-ID"),
+    user_id: str = Depends(get_current_user_id),
     service: MemoUseCase = Depends(get_memo_use_case),
 ):
     memo = service.get_memo_by_id(
         memo_id=memo_id,
-        user_id=x_user_id,
+        user_id=user_id,
     )
 
     if not memo:
@@ -88,13 +121,13 @@ def get_memo(
 def update_memo(
     memo_id: str,
     request: UpdateMemoRequest,
-    x_user_id: str = Header(..., alias="X-USER-ID"),
+    user_id: str = Depends(get_current_user_id),
     service: MemoUseCase = Depends(get_memo_use_case),
 ):
     try:
         updated_memo = service.update_memo(
             memo_id=memo_id,
-            user_id=x_user_id,
+            user_id=user_id,
             content=request.content,
         )
         return updated_memo
@@ -109,11 +142,11 @@ def update_memo(
 @router.delete("/memos/{memo_id}")
 def delete_memo(
     memo_id: str,
-    x_user_id: str = Header(..., alias="X-USER-ID"),
+    user_id: str = Depends(get_current_user_id),
     service: MemoUseCase = Depends(get_memo_use_case),
 ):
     try:
-        service.delete_memo(memo_id=memo_id, user_id=x_user_id)
+        service.delete_memo(memo_id=memo_id, user_id=user_id)
         return {"message": "메모 삭제 완료"}
     except MemoAccessDeniedError as e:
         raise HTTPException(status_code=403, detail=str(e))

backend/dependencies.py

@@ -0,0 +1,14 @@
+import os
+
+from backend.adapters.outbound.dynamodb_repository import DynamoDBRepository
+from backend.application.ports import MemoUseCase
+from backend.application.service import MemoService
+
+
+def get_memo_use_case() -> MemoUseCase:
+    table_name = os.getenv("DYNAMODB_TABLE")
+    if not table_name:
+        raise ValueError("DYNAMODB_TABLE 환경 변수가 설정되지 않았습니다.")
+
+    repo = DynamoDBRepository(table_name=table_name)
+    return MemoService(repo)

cli/adapters/inbound/cli_adapter.py

@@ -0,0 +1,149 @@
+import argparse
+
+from cli.application.exceptions import (
+    AuthDeviceFlowError,
+    AuthenticationError,
+)
+from cli.dependencies import container
+
+
+def main():
+    parser = argparse.ArgumentParser(description="mylog CLI", add_help=False)
+    parser.add_argument("-h", "--help", action="store_true", help="도움말 출력")
+
+    subparsers = parser.add_subparsers(dest="command")
+
+    subparsers.add_parser("login", help="깃허브 계정으로 로그인합니다.")
+    subparsers.add_parser("logout", help="로그아웃합니다.")
+
+    add_parser = subparsers.add_parser("add", help="새로운 메모를 기록합니다.")
+    add_parser.add_argument("content", type=str, help="저장할 메모 내용")
+    add_parser.add_argument(
+        "-c", "--category", type=str, default=None, help="카테고리 지정"
+    )
+
+    list_parser = subparsers.add_parser("list", help="저장된 메모를 조회합니다.")
+    list_parser.add_argument(
+        "category", type=str, nargs="?", default=None, help="조회할 카테고리"
+    )
+    list_parser.add_argument(
+        "-l", "--limit", type=int, default=5, help="출력 개수 지정"
+    )
+
+    update_parser = subparsers.add_parser("update", help="기존 메모를 수정합니다.")
+    update_parser.add_argument("id", type=str, help="수정할 메모 고유 ID")
+    update_parser.add_argument("content", type=str, help="새로운 내용")
+
+    rm_parser = subparsers.add_parser("rm", help="메모를 삭제합니다.")
+    rm_parser.add_argument("id", type=str, help="삭제할 메모 고유 ID")
+
+    subparsers.add_parser("help", help="도움말 매뉴얼을 출력합니다.")
+
+    args, unknown = parser.parse_known_args()
+
+    if args.help or args.command == "help" or args.command is None:
+        print("""Usage (사용 예시):
+
+mylog [options]
+
+Commands:
+login      깃허브 계정으로 로그인하여 클라우드 환경을 연동합니다.
+add        새로운 메모를 기록합니다. (-c 카테고리 지정)
+list       저장된 메모를 조회합니다. (-l 개수)
+update     기존 메모의 내용을 수정합니다. (ID 필요)
+rm         저장된 메모를 삭제합니다. (ID 필요)
+help       현재 화면의 도움말 매뉴얼을 출력합니다.""")
+        return
+
+    if args.command == "login":
+        try:
+            print("GitHub 서버에 로그인 요청 코드를 생성하는 중...")
+            login_data = container.auth_service.request_login()
+
+            print("\n" + "=" * 60)
+            print("[GitHub 로그인 안내]")
+            print("1. 웹 브라우저를 열고 다음 주소로 접속하세요.")
+            print(f"   {login_data['verification_uri']}")
+            print("2. 아래 코드를 입력하고 승인을 완료하세요.")
+            print(f"   [ {login_data['user_code']} ]")
+            print("=" * 60 + "\n")
+            print("사용자의 브라우저 로그인을 기다리는 중입니다...")
+
+            if container.auth_service.complete_login(
+                login_data["device_code"], login_data["interval"]
+            ):
+                print("\n[로그인 성공]")
+                print("깃허브 토큰이 안전하게 암호화 되어 저장되었습니다.")
+            else:
+                print("\n[로그인 실패] 깃허브 토큰을 획득하지 못했습니다.")
+        except AuthDeviceFlowError as e:
+            print(f"\n[로그인 실패] {e}")
+        except Exception as e:
+            print(f"로그인 예외 발생: {e}")
+
+    elif args.command == "logout":
+        try:
+            if container.auth_service.logout():
+                print("토큰이 안전하게 파기되었습니다. (로그아웃 성공)")
+            else:
+                print("이미 로그아웃된 상태거나 삭제할 토큰이 없습니다.")
+        except Exception as e:
+            print(f"로그아웃 실패: {e}")
+
+    elif args.command == "add":
+        try:
+            container.memo_service.save_memo(
+                content=args.content,
+                category=args.category,
+            )
+            cat_title = args.category if args.category else "basic"
+            print(f"\n> [{cat_title}] 메모가 저장되었습니다.")
+        except AuthenticationError as e:
+            print(f"\n[인증 실패] {e}")
+        except Exception as e:
+            print(f"\n[저장 실패] {e}")
+
+    elif args.command == "list":
+        try:
+            memos = container.memo_service.get_memos(
+                category=args.category,
+                limit=args.limit,
+            )
+
+            cat_title = args.category if args.category else "전체"
+            print(f"\n> [{cat_title}] 최근 메모 ({len(memos)}개 출력)")
+            for m in memos:
+                print(
+                    f"  [ID: {m.get('id', 'N/A')}] "
+                    f"{m.get('created_at', '')} | "
+                    f"{m.get('content', '')}"
+                )
+        except AuthenticationError as e:
+            print(f"\n[인증 실패] {e}")
+        except Exception as e:
+            print(f"\n[조회 실패] {e}")
+
+    elif args.command == "update":
+        try:
+            container.memo_service.update_memo(
+                memo_id=args.id,
+                content=args.content,
+            )
+            print(f"\n> [{args.id}] 메모가 성공적으로 수정되었습니다.")
+        except AuthenticationError as e:
+            print(f"\n[인증 실패] {e}")
+        except Exception as e:
+            print(f"\n[수정 실패] {e}")
+
+    elif args.command == "rm":
+        try:
+            container.memo_service.delete_memo(memo_id=args.id)
+            print(f"\n> {args.id}번 메모가 깔끔하게 삭제되었습니다.")
+        except AuthenticationError as e:
+            print(f"\n[인증 실패] {e}")
+        except Exception as e:
+            print(f"\n[삭제 실패] {e}")
+
+
+if __name__ == "__main__":
+    main()

cli/adapters/outbound/github_auth_adapter.py

@@ -0,0 +1,51 @@
+import time
+
+import requests
+
+from cli.application.ports.outbound import AuthRepositoryPort
+
+
+class GitHubDeviceAuthAdapter(AuthRepositoryPort):
+    DEVICE_CODE_URL = "https://github.com/login/device/code"
+    ACCESS_TOKEN_URL = "https://github.com/login/oauth/access_token"
+
+    def __init__(self, client_id: str):
+        self.client_id = client_id
+
+    def request_device_code(self) -> dict:
+        headers = {"Accept": "application/json"}
+        payload = {
+            "client_id": self.client_id,
+            "scope": "read:user",
+        }
+        response = requests.post(
+            self.DEVICE_CODE_URL, json=payload, headers=headers, timeout=5
+        )
+        response.raise_for_status()
+        return response.json()
+
+    def poll_for_token(self, device_code: str, interval: int = 5) -> dict:
+        headers = {"Accept": "application/json"}
+        payload = {
+            "client_id": self.client_id,
+            "device_code": device_code,
+            "grant_type": "urn:ietf:params:oauth:grant-type:device_code",
+        }
+
+        while True:
+            time.sleep(interval)
+            response = requests.post(
+                self.ACCESS_TOKEN_URL, json=payload, headers=headers, timeout=5
+            )
+            data = response.json()
+
+            if "access_token" in data:
+                return data
+
+            error = data.get("error")
+            if error == "authorization_pending":
+                continue
+            elif error == "slow_down":
+                interval += 5
+            else:
+                raise RuntimeError(f"Github 인증 실패: {error}")