refactor: repository -> service -> api & 의존성 주입

Author: nahyeongjin1

app/api/api_key_api.py

@@ -1,10 +1,12 @@
-from fastapi import APIRouter
+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 import api_key_service
+from app.services.api_key_service import APIKeyService, api_key_service
+
+api_key_service_dependency = Depends(lambda: api_key_service)
 
 router = APIRouter()
 
@@ -15,12 +17,14 @@
     summary="API KEY 저장 (처음 한 번)",
     description="외부 AI 서비스의 API Key를 암호화하여 로컬 데이터베이스에 저장합니다.",
 )
-def store_api_key(credential: APIKeyCreate) -> ResponseMessage[APIKeyResponse]:
+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 = api_key_service.store_api_key(credential)
+    created_credential = service.store_api_key(credential)
 
     response_data = APIKeyResponse(
         id=created_credential.id,

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

@@ -1,4 +1,3 @@
-# app/schemas/api_key/base_model.py
 from pydantic import BaseModel, Field
 
 from app.core.enum.llm_service import LLMServiceEnum

app/schemas/api_key/create_model.py

@@ -1,4 +1,3 @@
-# app/schemas/api_key/create_model.py
 from pydantic import Field, field_validator
 
 from app.schemas.api_key.base_model import APIKeyBase

app/schemas/api_key/db_model.py

@@ -1,4 +1,3 @@
-# app/schemas/api_key/db_model.py
 from datetime import datetime
 
 from app.schemas.api_key.base_model import APIKeyBase

app/schemas/api_key/response_model.py

@@ -1,4 +1,3 @@
-# app/schemas/api_key/response_model.py
 from datetime import datetime
 
 from pydantic import Field

app/services/api_key_service.py

@@ -1,53 +1,47 @@
 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, get_db_path
+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를 통해 데이터베이스에 저장합니다."""
+        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
+
 
-def store_api_key(credential_data: APIKeyCreate) -> APIKeyInDB:
-    """API_KEY를 암호화하여 데이터베이스에 저장합니다."""
-
-    encrypted_key = AES256.encrypt(credential_data.api_key)
-    new_id = generate_prefixed_uuid("QGENIE")
-
-    db_path = get_db_path()
-    conn = None
-    try:
-        # timeout을 10초로 설정하여 BUSY 상태에서 대기하도록 함
-        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, credential_data.service_name, encrypted_key),
-        )
-        conn.commit()
-
-        cursor.execute("SELECT * FROM ai_credential WHERE id = ?", (new_id,))
-        created_row = cursor.fetchone()
-
-        if not created_row:
-            raise APIException(CommonCode.FAIL_TO_VERIFY_CREATION)
-
-        return APIKeyInDB.model_validate(dict(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
-    finally:
-        if conn:
-            conn.close()
+api_key_service = APIKeyService()