디비 커넥션 작업

app/api/api_router.py

@@ -1,11 +1,14 @@
+# app/api/api_router.py
+
 from fastapi import APIRouter
 
-from app.api import driver_api, test_api
+from app.api import driver_api, test_api, user_db_api
 
 api_router = APIRouter()
 
 # 테스트 라우터
 api_router.include_router(test_api.router, prefix="/test", tags=["Test"])
 
 # 라우터
-api_router.include_router(driver_api.router, prefix="/connections", tags=["Driver"])
+api_router.include_router(driver_api.router, prefix="/driver", tags=["Driver"])
+api_router.include_router(user_db_api.router, prefix="/user/db", tags=["UserDb"])

app/api/driver_api.py

@@ -1,24 +1,34 @@
 # app/api/driver_api.py
 
-from fastapi import APIRouter
+from fastapi import APIRouter, Depends
 
 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.services.driver_service import db_driver_info
+from app.schemas.driver.driver_info_model import DriverInfo
+from app.services.driver_service import DriverService, driver_service
+
+driver_service_dependency = Depends(lambda: driver_service)
 
 router = APIRouter()
 
 
-@router.get("/drivers/{driverId}", response_model=ResponseMessage[DriverInfo], summary="DB 드라이버 정보 조회 API")
-def read_driver_info(driverId: str):
-    """DB 드라이버 정보 조회"""
+@router.get(
+    "/info/{driver_id}",
+    response_model=ResponseMessage[DriverInfo],
+    summary="DB 드라이버 정보 조회",
+)
+def read_driver_info(
+    driver_id: str,
+    service: DriverService = driver_service_dependency,
+) -> ResponseMessage[DriverInfo]:
+    """경로 파라미터로 받은 driver_id에 해당하는 DB 드라이버의 지원 정보를 조회합니다."""
     try:
-        # DBTypesEnum에서 driverID에 맞는 객체를 가져옵니다.
-        db_type_enum = DBTypesEnum[driverId.lower()]
-        return ResponseMessage.success(value=db_driver_info(DriverInfo.from_enum(db_type_enum)))
-    # db_type_enum 유효성 검사 실패
-    except KeyError:
-        raise APIException(CommonCode.INVALID_ENUM_VALUE) from KeyError
+        db_type_enum = DBTypesEnum[driver_id.lower()]
+    except KeyError as e:
+        raise APIException(CommonCode.INVALID_DB_DRIVER, *e.args) from e
+    driver_info_data = DriverInfo.from_enum(db_type_enum)
+    return ResponseMessage.success(
+        value=service.read_driver_info(driver_info_data), code=CommonCode.SUCCESS_DRIVER_INFO
+    )

app/api/health.py → app/api/health_api.py

@@ -1,4 +1,4 @@
-# app/api/health.py
+# app/api/health_api.py
 from fastapi import APIRouter
 
 router = APIRouter(tags=["Health"])

app/api/user_db_api.py

@@ -0,0 +1,30 @@
+# app/api/user_db_api.py
+
+from fastapi import APIRouter, Depends
+
+from app.core.exceptions import APIException
+from app.core.response import ResponseMessage
+from app.schemas.user_db.db_profile_model import DBProfileCreate
+from app.services.user_db_service import UserDbService, user_db_service
+
+user_db_service_dependency = Depends(lambda: user_db_service)
+
+router = APIRouter()
+
+
+@router.post(
+    "/connect/test",
+    response_model=ResponseMessage[bool],
+    summary="DB 연결 테스트",
+)
+def connection_test(
+    db_info: DBProfileCreate,
+    service: UserDbService = user_db_service_dependency,
+) -> ResponseMessage[bool]:
+    """DB 연결 정보를 받아 연결 가능 여부를 테스트합니다."""
+    db_info.validate_required_fields()
+
+    result = service.connection_test(db_info)
+    if not result.is_successful:
+        raise APIException(result.code)
+    return ResponseMessage.success(value=result.is_successful, code=result.code)

app/core/enum/db_driver.py

@@ -8,6 +8,6 @@ class DBTypesEnum(Enum):
     postgresql = "psycopg2"
     mysql = "mysql.connector"
     sqlite = "sqlite3"
-    oracle = "cx_Oracle"
+    oracle = "oracledb"
     sqlserver = "pyodbc"
     mariadb = "pymysql"

app/core/exceptions.py

@@ -1,45 +1,68 @@
 import traceback
+from typing import Any
 
 from fastapi import Request
+from fastapi.exceptions import RequestValidationError
 from fastapi.responses import JSONResponse
 
 from app.core.status import CommonCode
 
 
+def _create_error_response(code: CommonCode, data: Any | None = None) -> JSONResponse:
+    """
+    모든 에러 응답에 사용될 표준 JSONResponse 객체를 생성하는 헬퍼 함수.
+    """
+    error_content = {
+        "code": code.code,
+        "message": code.message,
+        "data": data,
+    }
+    return JSONResponse(
+        status_code=code.http_status,
+        content=error_content,
+    )
+
+
 class APIException(Exception):
     """
     API 로직 내에서 발생하는 모든 예상된 오류에 사용할 기본 예외 클래스입니다.
     """
 
     def __init__(self, code: CommonCode, *args):
         self.code_enum = code
-        self.message = code.get_message(*args)
+        self.message = code.message
+        self.args = args
         super().__init__(self.message)
 
 
+async def validation_exception_handler(request: Request, exc: RequestValidationError):
+    """
+    Pydantic 모델의 유효성 검사 실패(RequestValidationError)를 감지하여
+    표준화된 JSON 오류 응답을 반환합니다.
+    """
+    error_details = []
+    for error in exc.errors():
+        field_name = ".".join(map(str, error["loc"][1:]))
+        error_details.append({"field": field_name, "message": error["msg"]})
+
+    return _create_error_response(code=CommonCode.INVALID_PARAMETER, data={"details": error_details})
+
+
 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},
-    )
+    return _create_error_response(code=exc.code_enum, data=exc.args)
 
 
 async def generic_exception_handler(request: Request, exc: Exception):
     """
     처리되지 않은 모든 예외를 잡아, 일관된 500 서버 오류를 반환합니다.
     """
-    # 운영 환경에서는 파일 로그나 모니터링 시스템으로 보내야 합니다.
+    error_traceback = traceback.format_exc()
+
     print("=" * 20, "UNEXPECTED ERROR", "=" * 20)
-    traceback.print_exc()
+    print(error_traceback)
     print("=" * 50)
 
-    # 사용자에게는 간단한 500 에러 메시지만 보여줍니다.
-    error_response = {"code": CommonCode.FAIL.code, "message": CommonCode.FAIL.message, "data": None}
-
-    return JSONResponse(
-        status_code=CommonCode.FAIL.http_status,
-        content=error_response,
-    )
+    return _create_error_response(code=CommonCode.FAIL, data={"traceback": error_traceback})

app/core/status.py

@@ -10,25 +10,62 @@ class CommonCode(Enum):
     상태 코드 참고: https://developer.mozilla.org/ko/docs/Web/HTTP/Status
     """
 
-    # ==================================
-    #    성공 (Success) - 2xx
-    # ==================================
+    # =======================================
+    #    성공 (Success) - 2xxx
+    # =======================================
+    """ 기본 성공 코드 - 20xx """
     SUCCESS = (status.HTTP_200_OK, "2000", "성공적으로 처리되었습니다.")
     CREATED = (status.HTTP_201_CREATED, "2001", "성공적으로 생성되었습니다.")
-    SUCCESS_DB_CONNECT = (status.HTTP_200_OK, "2002", "디비 연결을 성공하였습니다.")
 
-    # ==================================
-    #    클라이언트 오류 (Client Error) - 4xx
-    # ==================================
+    """ DRIVER, DB 성공 코드 - 21xx """
+    SUCCESS_DRIVER_INFO = (status.HTTP_200_OK, "2100", "드라이버 정보 조회를 성공하였습니다.")
+    SUCCESS_USER_DB_CONNECT_TEST = (status.HTTP_200_OK, "2101", "테스트 연결을 성공하였습니다.")
+
+    """ KEY 성공 코드 - 22xx """
+
+    """ AI CHAT, DB 성공 코드 - 23xx """
+
+    """ ANNOTATION 성공 코드 - 24xx """
+
+    """ SQL 성공 코드 - 25xx """
+
+    # =======================================
+    #    클라이언트 오류 (Client Error) - 4xxx
+    # =======================================
+    """ 기본 클라이언트 오류 코드 - 40xx """
     NO_VALUE = (status.HTTP_400_BAD_REQUEST, "4000", "필수 값이 존재하지 않습니다.")
     DUPLICATION = (status.HTTP_409_CONFLICT, "4001", "이미 존재하는 데이터입니다.")
     NO_SEARCH_DATA = (status.HTTP_404_NOT_FOUND, "4002", "요청한 데이터를 찾을 수 없습니다.")
-    INVALID_ENUM_VALUE = (status.HTTP_422_UNPROCESSABLE_ENTITY, "4003", "지원하지 않는 데이터베이스 값입니다.")
+    INVALID_PARAMETER = (status.HTTP_422_UNPROCESSABLE_ENTITY, "4003", "필수 값이 누락되었습니다.")
+
+    """ DRIVER, DB 클라이언트 오류 코드 - 41xx """
+    INVALID_DB_DRIVER = (status.HTTP_409_CONFLICT, "4100", "지원하지 않는 데이터베이스입니다.")
+    NO_DB_DRIVER = (status.HTTP_400_BAD_REQUEST, "4101", "데이터베이스는 필수 값입니다.")
+
+    """ KEY 클라이언트 오류 코드 - 42xx """
+
+    """ AI CHAT, DB 클라이언트 오류 코드 - 43xx """
+
+    """ ANNOTATION 클라이언트 오류 코드 - 44xx """
+
+    """ SQL 클라이언트 오류 코드 - 45xx """
 
     # ==================================
     #    서버 오류 (Server Error) - 5xx
     # ==================================
-    FAIL = (status.HTTP_500_INTERNAL_SERVER_ERROR, "9999", "서버 처리 중 오류가 발생했습니다.")
+    """ 기본 서버 오류 코드 - 50xx """
+    FAIL = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5000", "서버 처리 중 오류가 발생했습니다.")
+
+    """ DRIVER, DB 서버 오류 코드 - 51xx """
+    FAIL_CONNECT_DB = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5100", "디비 연결 중 오류가 발생했습니다.")
+
+    """ KEY 서버 오류 코드 - 52xx """
+
+    """ AI CHAT, DB 서버 오류 코드 - 53xx """
+
+    """ ANNOTATION 서버 오류 코드 - 54xx """
+
+    """ SQL 서버 오류 코드 - 55xx """
 
     def __init__(self, http_status: int, code: str, message: str):
         """Enum 멤버가 생성될 때 각 값을 속성으로 할당합니다."""
@@ -40,4 +77,7 @@ def get_message(self, *args) -> str:
         """
         메시지 포맷팅이 필요한 경우, 인자를 받아 완성된 메시지를 반환합니다.
         """
-        return self.message % args if args else self.message
+        try:
+            return self.message % args if args else self.message
+        except Exception:
+            return self.message

app/main.py

@@ -2,20 +2,27 @@
 
 import uvicorn
 from fastapi import FastAPI
+from fastapi.exceptions import RequestValidationError
 
-from app.api import health  # 헬스 체크
+from app.api import health_api
 from app.api.api_router import api_router
-from app.core.exceptions import APIException, api_exception_handler, generic_exception_handler
+from app.core.exceptions import (
+    APIException,
+    api_exception_handler,
+    generic_exception_handler,
+    validation_exception_handler,
+)
 from app.db.init_db import initialize_database
 
 app = FastAPI()
 
 # 전역 예외 처리기 등록
-app.add_exception_handler(APIException, api_exception_handler)
 app.add_exception_handler(Exception, generic_exception_handler)
+app.add_exception_handler(APIException, api_exception_handler)
+app.add_exception_handler(RequestValidationError, validation_exception_handler)
 
 # 라우터
-app.include_router(health.router)
+app.include_router(health_api.router)
 app.include_router(api_router, prefix="/api")
 
 # initialize_database 함수가 호출되어 테이블이 생성되거나 이미 존재함을 확인합니다.

app/repository/user_db_repository.py

@@ -0,0 +1,39 @@
+from typing import Any
+
+import oracledb
+
+from app.core.status import CommonCode
+from app.schemas.user_db.connect_test_result_model import TestConnectionResult
+
+
+class UserDbRepository:
+    def test_db_connection(self, driver_module: Any, **kwargs: Any) -> TestConnectionResult:
+        """
+        DB 드라이버와 연결에 필요한 매개변수들을 받아 연결을 테스트합니다.
+        """
+        connection = None
+        try:
+            if driver_module is oracledb:
+                if kwargs.get("user").lower() == "sys":
+                    kwargs["mode"] = oracledb.AUTH_MODE_SYSDBA
+                connection = driver_module.connect(**kwargs)
+            # MSSQL과 같이 전체 연결 문자열이 제공된 경우
+            elif "connection_string" in kwargs:
+                connection = driver_module.connect(kwargs["connection_string"])
+            # SQLite와 같이 파일 이름만 필요한 경우
+            elif "db_name" in kwargs:
+                connection = driver_module.connect(kwargs["db_name"])
+            # 그 외 (MySQL, PostgreSQL, Oracle 등) 일반적인 키워드 인자 방식 연결
+            else:
+                connection = driver_module.connect(**kwargs)
+
+            return TestConnectionResult(is_successful=True, code=CommonCode.SUCCESS_USER_DB_CONNECT_TEST)
+
+        except Exception:
+            return TestConnectionResult(is_successful=False, code=CommonCode.FAIL_CONNECT_DB)
+        finally:
+            if connection:
+                connection.close()
+
+
+user_db_repository = UserDbRepository()

app/schemas/driver_info.py → app/schemas/driver/driver_info_model.py

@@ -1,4 +1,4 @@
-# app/schemas/driver_info.py
+# app/schemas/driver/driver_info_model.py
 from pydantic import BaseModel
 
 from app.core.enum.db_driver import DBTypesEnum

app/schemas/user_db/connect_test_result_model.py

@@ -0,0 +1,10 @@
+# app/schemas/user_db/connect_test_result_model.py
+
+from pydantic import BaseModel, Field
+
+from app.core.status import CommonCode
+
+
+class TestConnectionResult(BaseModel):
+    is_successful: bool = Field(..., description="성공 여부")
+    code: CommonCode = Field(None, description="결과 코드")