diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 5977fee..99ba938 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -2,7 +2,7 @@ repos: - repo: https://github.com/astral-sh/ruff-pre-commit rev: "v0.12.2" # ruff 버전에 맞게 수정 hooks: - - id: ruff + - id: ruff-check args: ["--fix"] # 자동 수정 적용 - repo: https://github.com/psf/black diff --git a/app/api/driver_api.py b/app/api/driver_api.py index 8d488fc..bf5dbad 100644 --- a/app/api/driver_api.py +++ b/app/api/driver_api.py @@ -4,9 +4,9 @@ from app.core.enum.db_driver import DBTypesEnum from app.core.exceptions import APIException +from app.core.response import ResponseMessage from app.core.status import CommonCode from app.schemas.driver_info import DriverInfo -from app.core.response import ResponseMessage from app.services.driver_service import db_driver_info router = APIRouter() @@ -21,4 +21,4 @@ def read_driver_info(driverId: str): return ResponseMessage.success(value=db_driver_info(DriverInfo.from_enum(db_type_enum))) # db_type_enum 유효성 검사 실패 except KeyError: - raise APIException(CommonCode.INVALID_ENUM_VALUE) + raise APIException(CommonCode.INVALID_ENUM_VALUE) from KeyError diff --git a/app/api/test_api.py b/app/api/test_api.py index ca0dcca..49672b9 100644 --- a/app/api/test_api.py +++ b/app/api/test_api.py @@ -1,11 +1,12 @@ from fastapi import APIRouter -from app.core.response import ResponseMessage from app.core.exceptions import APIException +from app.core.response import ResponseMessage from app.core.status import CommonCode router = APIRouter() + @router.get("", response_model=ResponseMessage, summary="타입 변환을 이용한 성공/실패/버그 테스트") def simple_test(mode: str): """ @@ -27,14 +28,11 @@ def simple_test(mode: str): # 2. 정수로 변환 성공 시, 값에 따라 분기 if mode_int == 1: # 기본 성공 코드(SUCCESS)로 응답 - return ResponseMessage.success( - value={"detail": "기본 성공 테스트입니다."} - ) + return ResponseMessage.success(value={"detail": "기본 성공 테스트입니다."}) elif mode_int == 2: # 커스텀 성공 코드(CREATED)로 응답 return ResponseMessage.success( - value={"detail": "커스텀 성공 코드(CREATED) 테스트입니다."}, - code=CommonCode.CREATED + value={"detail": "커스텀 성공 코드(CREATED) 테스트입니다."}, code=CommonCode.CREATED ) else: # 그 외 숫자는 '데이터 없음' 오류로 처리 @@ -44,5 +42,4 @@ def simple_test(mode: str): # 3. 정수로 변환 실패 시 (문자열이 들어온 경우) # 예상치 못한 버그를 강제로 발생시킵니다. # 이 에러는 generic_exception_handler가 처리하게 됩니다. - raise TypeError("의도적으로 발생시킨 타입 에러입니다.") - + raise TypeError("의도적으로 발생시킨 타입 에러입니다.") from ValueError diff --git a/app/core/exceptions.py b/app/core/exceptions.py index ea68b2a..24f97de 100644 --- a/app/core/exceptions.py +++ b/app/core/exceptions.py @@ -1,6 +1,8 @@ import traceback -from fastapi import Request, status + +from fastapi import Request from fastapi.responses import JSONResponse + from app.core.status import CommonCode @@ -8,41 +10,36 @@ class APIException(Exception): """ API 로직 내에서 발생하는 모든 예상된 오류에 사용할 기본 예외 클래스입니다. """ + def __init__(self, code: CommonCode, *args): self.code_enum = code self.message = code.get_message(*args) super().__init__(self.message) + async def api_exception_handler(request: Request, exc: APIException): """ APIException이 발생했을 때, 이를 감지하여 표준화된 JSON 오류 응답을 반환합니다. """ return JSONResponse( status_code=exc.code_enum.http_status, - content={ - "code": exc.code_enum.code, - "message": exc.message, - "data": None - } + content={"code": exc.code_enum.code, "message": exc.message, "data": None}, ) + async def generic_exception_handler(request: Request, exc: Exception): """ 처리되지 않은 모든 예외를 잡아, 일관된 500 서버 오류를 반환합니다. """ # 운영 환경에서는 파일 로그나 모니터링 시스템으로 보내야 합니다. - print("="*20, "UNEXPECTED ERROR", "="*20) + print("=" * 20, "UNEXPECTED ERROR", "=" * 20) traceback.print_exc() - print("="*50) + print("=" * 50) # 사용자에게는 간단한 500 에러 메시지만 보여줍니다. - error_response = { - "code": CommonCode.FAIL.code, - "message": CommonCode.FAIL.message, - "data": None - } + error_response = {"code": CommonCode.FAIL.code, "message": CommonCode.FAIL.message, "data": None} return JSONResponse( status_code=CommonCode.FAIL.http_status, content=error_response, - ) \ No newline at end of file + ) diff --git a/app/core/response.py b/app/core/response.py index c57d50a..ba7bf61 100644 --- a/app/core/response.py +++ b/app/core/response.py @@ -1,29 +1,24 @@ -from typing import Generic, TypeVar, Optional +from typing import Generic, TypeVar + from pydantic import BaseModel, Field + from app.core.status import CommonCode -T = TypeVar('T') +T = TypeVar("T") + class ResponseMessage(BaseModel, Generic[T]): """ 모든 API 응답에 사용될 공용 스키마입니다. """ + code: str = Field(..., description="응답을 나타내는 고유 상태 코드") message: str = Field(..., description="응답 메시지") - data: Optional[T] = Field(None, description="반환될 실제 데이터") + data: T | None = Field(None, description="반환될 실제 데이터") @classmethod - def success( - cls, - value: Optional[T] = None, - code: CommonCode = CommonCode.SUCCESS, - *args - ) -> "ResponseMessage[T]": + def success(cls, value: T | None = None, code: CommonCode = CommonCode.SUCCESS, *args) -> "ResponseMessage[T]": """ 성공 응답을 생성하는 팩토리 메서드입니다. """ - return cls( - code=code.code, - message=code.get_message(*args), - data=value - ) + return cls(code=code.code, message=code.get_message(*args), data=value) diff --git a/app/core/security.py b/app/core/security.py index 47376d7..d598030 100644 --- a/app/core/security.py +++ b/app/core/security.py @@ -1,13 +1,16 @@ -import os import base64 +import os + from Crypto.Cipher import AES -from Crypto.Util.Padding import pad, unpad from Crypto.Random import get_random_bytes +from Crypto.Util.Padding import pad, unpad """ 보안 원칙을 적용한 AES-256 암호화 및 복호화 클래스입니다. - 암호화 시 매번 새로운 랜덤 IV를 생성합니다. """ + + class AES256: _key = base64.b64decode(os.getenv("ENV_AES256_KEY")) @@ -17,13 +20,13 @@ def encrypt(text: str) -> str: cipher = AES.new(AES256._key, AES.MODE_CBC, iv) - data_bytes = text.encode('utf-8') + data_bytes = text.encode("utf-8") padded_bytes = pad(data_bytes, AES.block_size) encrypted_bytes = cipher.encrypt(padded_bytes) combined_bytes = iv + encrypted_bytes - return base64.b64encode(combined_bytes).decode('utf-8') + return base64.b64encode(combined_bytes).decode("utf-8") @staticmethod def decrypt(cipher_text: str) -> str: @@ -32,13 +35,12 @@ def decrypt(cipher_text: str) -> str: """ combined_bytes = base64.b64decode(cipher_text) - iv = combined_bytes[:AES.block_size] - encrypted_bytes = combined_bytes[AES.block_size:] + iv = combined_bytes[: AES.block_size] + encrypted_bytes = combined_bytes[AES.block_size :] cipher = AES.new(AES256._key, AES.MODE_CBC, iv) decrypted_padded_bytes = cipher.decrypt(encrypted_bytes) decrypted_bytes = unpad(decrypted_padded_bytes, AES.block_size) - return decrypted_bytes.decode('utf-8') - + return decrypted_bytes.decode("utf-8") diff --git a/app/db/init_db.py b/app/db/init_db.py index 4de856d..0b8fa58 100644 --- a/app/db/init_db.py +++ b/app/db/init_db.py @@ -1,10 +1,13 @@ # db/init_db.py import sqlite3 + from app.core.utils import get_db_path """ 데이터베이스에 연결하고, 애플리케이션에 필요한 테이블이 없으면 생성합니다. """ + + def initialize_database(): db_path = get_db_path() @@ -13,7 +16,8 @@ def initialize_database(): conn = sqlite3.connect(db_path) cursor = conn.cursor() # db_profile 테이블 생성 - cursor.execute(""" + cursor.execute( + """ CREATE TABLE IF NOT EXISTS db_profile ( id VARCHAR(64) PRIMARY KEY NOT NULL, type VARCHAR(32) NOT NULL, @@ -25,19 +29,23 @@ def initialize_database(): created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ); - """) + """ + ) # db_profile 테이블의 updated_at을 자동으로 업데이트하는 트리거 - cursor.execute(""" + cursor.execute( + """ CREATE TRIGGER IF NOT EXISTS update_db_profile_updated_at BEFORE UPDATE ON db_profile FOR EACH ROW BEGIN UPDATE db_profile SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; END; - """) + """ + ) # ai_credential 테이블 생성 - cursor.execute(""" + cursor.execute( + """ CREATE TABLE IF NOT EXISTS ai_credential ( id VARCHAR(64) PRIMARY KEY NOT NULL, service_name VARCHAR(32) NOT NULL, @@ -45,38 +53,46 @@ def initialize_database(): created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ); - """) + """ + ) # ai_credential 테이블의 updated_at을 자동으로 업데이트하는 트리거 - cursor.execute(""" + cursor.execute( + """ CREATE TRIGGER IF NOT EXISTS update_ai_credential_updated_at BEFORE UPDATE ON ai_credential FOR EACH ROW BEGIN UPDATE ai_credential SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; END; - """) + """ + ) # chat_tab 테이블 생성 - cursor.execute(""" + cursor.execute( + """ CREATE TABLE IF NOT EXISTS chat_tab ( id VARCHAR(64) PRIMARY KEY NOT NULL, name VARCHAR(255), created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ); - """) + """ + ) # chat_tab 테이블의 updated_at을 자동으로 업데이트하는 트리거 - cursor.execute(""" + cursor.execute( + """ CREATE TRIGGER IF NOT EXISTS update_chat_tab_updated_at BEFORE UPDATE ON chat_tab FOR EACH ROW BEGIN UPDATE chat_tab SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; END; - """) + """ + ) # chat_message 테이블 생성 - cursor.execute(""" + cursor.execute( + """ CREATE TABLE IF NOT EXISTS chat_message ( id VARCHAR(64) PRIMARY KEY NOT NULL, chat_tab_id VARCHAR(64) NOT NULL, @@ -86,19 +102,23 @@ def initialize_database(): updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (chat_tab_id) REFERENCES chat_tab(id) ); - """) + """ + ) # chat_message 테이블의 updated_at을 자동으로 업데이트하는 트리거 - cursor.execute(""" + cursor.execute( + """ CREATE TRIGGER IF NOT EXISTS update_chat_message_updated_at BEFORE UPDATE ON chat_message FOR EACH ROW BEGIN UPDATE chat_message SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; END; - """) + """ + ) # query_history 테이블 생성 - cursor.execute(""" + cursor.execute( + """ CREATE TABLE IF NOT EXISTS query_history ( id VARCHAR(64) PRIMARY KEY NOT NULL, chat_message_id VARCHAR(64) NOT NULL, @@ -109,16 +129,19 @@ def initialize_database(): updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (chat_message_id) REFERENCES chat_message(id) ); - """) + """ + ) # query_history 테이블의 updated_at을 자동으로 업데이트하는 트리거 - cursor.execute(""" + cursor.execute( + """ CREATE TRIGGER IF NOT EXISTS update_query_history_updated_at BEFORE UPDATE ON query_history FOR EACH ROW BEGIN UPDATE query_history SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; END; - """) + """ + ) conn.commit() @@ -127,4 +150,3 @@ def initialize_database(): finally: if conn: conn.close() - diff --git a/app/services/driver_service.py b/app/services/driver_service.py index afed25c..a3ade45 100644 --- a/app/services/driver_service.py +++ b/app/services/driver_service.py @@ -16,5 +16,5 @@ def db_driver_info(driver_info: DriverInfo): return driver_info.update_from_module(version, size) - except (ModuleNotFoundError, AttributeError, OSError): - raise APIException(CommonCode.FAIL) + except (ModuleNotFoundError, AttributeError, OSError) as e: + raise APIException(CommonCode.FAIL) from e