[BAK-30] [기능] API_KEY 저장

app/api/api_key_api.py

@@ -0,0 +1,37 @@
+from fastapi import APIRouter, Depends
+
+from app.core.response import ResponseMessage
+from app.core.status import CommonCode
+from app.schemas.api_key.create_model import APIKeyCreate
+from app.schemas.api_key.response_model import APIKeyResponse
+from app.services.api_key_service import APIKeyService, api_key_service
+
+api_key_service_dependency = Depends(lambda: api_key_service)
+
+router = APIRouter()
+
+
+@router.post(
+    "/actions",
+    response_model=ResponseMessage[APIKeyResponse],
+    summary="API KEY 저장 (처음 한 번)",
+    description="외부 AI 서비스의 API Key를 암호화하여 로컬 데이터베이스에 저장합니다.",
+)
+def store_api_key(
+    credential: APIKeyCreate, service: APIKeyService = api_key_service_dependency
+) -> ResponseMessage[APIKeyResponse]:
+    """
+    - **service_name**: API Key가 사용될 외부 서비스 이름 (예: "OpenAI")
+    - **api_key**: 암호화하여 저장할 실제 API Key (예: "sk-***..")
+    """
+    created_credential = service.store_api_key(credential)
+
+    response_data = APIKeyResponse(
+        id=created_credential.id,
+        service_name=created_credential.service_name.value,
+        api_key_encrypted=created_credential.api_key,
+        created_at=created_credential.created_at,
+        updated_at=created_credential.updated_at,
+    )
+
+    return ResponseMessage.success(value=response_data, code=CommonCode.CREATED)

app/api/api_router.py

@@ -2,7 +2,7 @@
 
 from fastapi import APIRouter
 
-from app.api import driver_api, test_api, user_db_api
+from app.api import api_key_api, driver_api, test_api, user_db_api
 
 api_router = APIRouter()
 
@@ -12,3 +12,4 @@
 # 라우터
 api_router.include_router(driver_api.router, prefix="/driver", tags=["Driver"])
 api_router.include_router(user_db_api.router, prefix="/user/db", tags=["UserDb"])
+api_router.include_router(api_key_api.router, prefix="/credentials", tags=["Credentials"])

app/core/enum/llm_service_info.py

@@ -0,0 +1,10 @@
+from enum import Enum
+
+
+class LLMServiceEnum(str, Enum):
+    """지원하는 외부 LLM 서비스 목록"""
+
+    OPENAI = "OpenAI"
+    ANTHROPIC = "Anthropic"
+    GEMINI = "Gemini"
+    # TODO: 다른 지원 서비스를 여기에 추가

app/core/status.py

@@ -43,6 +43,12 @@ class CommonCode(Enum):
     NO_DB_DRIVER = (status.HTTP_400_BAD_REQUEST, "4101", "데이터베이스는 필수 값입니다.")
 
     """ KEY 클라이언트 오류 코드 - 42xx """
+    INVALID_API_KEY_FORMAT = (status.HTTP_400_BAD_REQUEST, "4200", "API 키의 형식이 올바르지 않습니다.")
+    INVALID_API_KEY_PREFIX = (
+        status.HTTP_400_BAD_REQUEST,
+        "4201",
+        "API 키가 선택한 서비스의 올바른 형식이 아닙니다. (예: OpenAI는 sk-로 시작)",
+    )
 
     """ AI CHAT, DB 클라이언트 오류 코드 - 43xx """
 
@@ -55,6 +61,16 @@ class CommonCode(Enum):
     # ==================================
     """ 기본 서버 오류 코드 - 50xx """
     FAIL = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5000", "서버 처리 중 오류가 발생했습니다.")
+    DB_BUSY = (
+        status.HTTP_503_SERVICE_UNAVAILABLE,
+        "5001",
+        "데이터베이스가 현재 사용 중입니다. 잠시 후 다시 시도해주세요.",
+    )
+    FAIL_TO_VERIFY_CREATION = (
+        status.HTTP_500_INTERNAL_SERVER_ERROR,
+        "5002",
+        "데이터 생성 후 검증 과정에서 오류가 발생했습니다.",
+    )
 
     """ DRIVER, DB 서버 오류 코드 - 51xx """
     FAIL_CONNECT_DB = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5100", "디비 연결 중 오류가 발생했습니다.")

app/db/init_db.py

@@ -48,7 +48,7 @@ def initialize_database():
             """
             CREATE TABLE IF NOT EXISTS ai_credential (
                 id VARCHAR(64) PRIMARY KEY NOT NULL,
-                service_name VARCHAR(32) NOT NULL,
+                service_name VARCHAR(32) NOT NULL UNIQUE,
                 api_key VARCHAR(256) NOT NULL,
                 created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
                 updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP

app/repository/api_key_repository.py

@@ -0,0 +1,42 @@
+import sqlite3
+
+from app.core.utils import get_db_path
+from app.schemas.api_key.db_model import APIKeyInDB
+
+
+class APIKeyRepository:
+    def create_api_key(self, new_id: str, service_name: str, encrypted_key: str) -> APIKeyInDB:
+        """
+        암호화된 API Key 정보를 받아 데이터베이스에 저장하고,
+        저장된 객체를 반환합니다.
+        """
+        db_path = get_db_path()
+        conn = None
+        try:
+            conn = sqlite3.connect(str(db_path), timeout=10)
+            conn.row_factory = sqlite3.Row
+            cursor = conn.cursor()
+
+            cursor.execute(
+                """
+                INSERT INTO ai_credential (id, service_name, api_key)
+                VALUES (?, ?, ?)
+                """,
+                (new_id, service_name, encrypted_key),
+            )
+            conn.commit()
+
+            cursor.execute("SELECT * FROM ai_credential WHERE id = ?", (new_id,))
+            created_row = cursor.fetchone()
+
+            if not created_row:
+                return None
+
+            return APIKeyInDB.model_validate(dict(created_row))
+
+        finally:
+            if conn:
+                conn.close()
+
+
+api_key_repository = APIKeyRepository()

app/schemas/api_key/base_model.py

@@ -0,0 +1,9 @@
+from pydantic import BaseModel, Field
+
+from app.core.enum.llm_service_info import LLMServiceEnum
+
+
+class APIKeyBase(BaseModel):
+    """API Key 도메인의 모든 스키마가 상속하는 기본 모델"""
+
+    service_name: LLMServiceEnum = Field(..., description="외부 서비스 이름")

app/schemas/api_key/create_model.py

@@ -0,0 +1,27 @@
+from pydantic import Field
+
+from app.core.enum.llm_service_info import LLMServiceEnum
+from app.core.exceptions import APIException
+from app.core.status import CommonCode
+from app.schemas.api_key.base_model import APIKeyBase
+
+
+class APIKeyCreate(APIKeyBase):
+    """API Key 생성을 위한 스키마"""
+
+    api_key: str = Field(..., description="암호화하여 저장할 실제 API Key")
+
+    def validate_with_service(self) -> None:
+        """서비스 종류에 따라 API Key의 유효성을 검증합니다."""
+        # 1. 기본 형식 검증 (공백 또는 빈 문자열)
+        if not self.api_key or self.api_key.isspace():
+            raise APIException(CommonCode.INVALID_API_KEY_FORMAT)
+
+        # 2. 서비스별 접두사 검증
+        key_prefix_map = {
+            LLMServiceEnum.OPENAI: "sk-",
+        }
+        required_prefix = key_prefix_map.get(self.service_name)
+
+        if required_prefix and not self.api_key.startswith(required_prefix):
+            raise APIException(CommonCode.INVALID_API_KEY_PREFIX)

app/schemas/api_key/db_model.py

@@ -0,0 +1,15 @@
+from datetime import datetime
+
+from app.schemas.api_key.base_model import APIKeyBase
+
+
+class APIKeyInDB(APIKeyBase):
+    """데이터베이스에 저장된 형태의 스키마 (내부용)"""
+
+    id: str
+    api_key: str  # DB 모델에서는 암호화된 키를 의미
+    created_at: datetime
+    updated_at: datetime
+
+    class Config:
+        from_attributes = True

app/schemas/api_key/response_model.py

@@ -0,0 +1,14 @@
+from datetime import datetime
+
+from pydantic import Field
+
+from app.schemas.api_key.base_model import APIKeyBase
+
+
+class APIKeyResponse(APIKeyBase):
+    """API 응답용 스키마"""
+
+    id: str
+    api_key_encrypted: str = Field(..., description="암호화된 API Key")
+    created_at: datetime
+    updated_at: datetime

app/services/api_key_service.py

@@ -0,0 +1,48 @@
+import sqlite3
+
+from fastapi import Depends
+
+from app.core.exceptions import APIException
+from app.core.security import AES256
+from app.core.status import CommonCode
+from app.core.utils import generate_prefixed_uuid
+from app.repository.api_key_repository import APIKeyRepository, api_key_repository
+from app.schemas.api_key.create_model import APIKeyCreate
+from app.schemas.api_key.db_model import APIKeyInDB
+
+api_key_repository_dependency = Depends(lambda: api_key_repository)
+
+
+class APIKeyService:
+    def store_api_key(
+        self, credential_data: APIKeyCreate, repository: APIKeyRepository = api_key_repository
+    ) -> APIKeyInDB:
+        """API_KEY를 암호화하고 repository를 통해 데이터베이스에 저장합니다."""
+        credential_data.validate_with_service()
+        try:
+            encrypted_key = AES256.encrypt(credential_data.api_key)
+            new_id = generate_prefixed_uuid("QGENIE")
+
+            created_row = repository.create_api_key(
+                new_id=new_id,
+                service_name=credential_data.service_name.value,
+                encrypted_key=encrypted_key,
+            )
+
+            if not created_row:
+                raise APIException(CommonCode.FAIL_TO_VERIFY_CREATION)
+
+            return created_row
+
+        except sqlite3.IntegrityError as e:
+            # UNIQUE 제약 조건 위반 (service_name)
+            raise APIException(CommonCode.DUPLICATION) from e
+        except sqlite3.Error as e:
+            # "database is locked" 오류를 명시적으로 처리
+            if "database is locked" in str(e):
+                raise APIException(CommonCode.DB_BUSY) from e
+            # 기타 모든 sqlite3 오류
+            raise APIException(CommonCode.FAIL) from e
+
+
+api_key_service = APIKeyService()