From 365c9278dafeeb42d780d0f3b760d6bc062cc2ef Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sat, 9 Aug 2025 00:12:28 +0900
Subject: [PATCH 01/25] =?UTF-8?q?feat:=20db=20key=20prefix=20=EC=9D=B4?=
 =?UTF-8?q?=EB=A6=84=EC=9D=84=20=EB=8B=B4=EA=B3=A0=EC=9E=88=EB=8A=94=20enu?=
 =?UTF-8?q?m=20=EC=83=9D=EC=84=B1?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/core/enum/db_key_prefix_name.py | 7 +++++++
 1 file changed, 7 insertions(+)
 create mode 100644 app/core/enum/db_key_prefix_name.py

diff --git a/app/core/enum/db_key_prefix_name.py b/app/core/enum/db_key_prefix_name.py
new file mode 100644
index 0000000..5eea582
--- /dev/null
+++ b/app/core/enum/db_key_prefix_name.py
@@ -0,0 +1,7 @@
+# app/core/enum/db_key_prefix_name.py
+from enum import Enum
+
+class DBSaveIdEnum(Enum):
+    """저장할 디비 ID 앞에 들어갈 이름"""
+    user_db = "USER-DB"
+    driver = "DRIVER"
\ No newline at end of file

From ca197f5af63ee13e82ccaed518ac2cf02d878443 Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sat, 9 Aug 2025 00:13:23 +0900
Subject: [PATCH 02/25] =?UTF-8?q?feat:=20db=20profile=20=EC=A0=80=EC=9E=A5?=
 =?UTF-8?q?=20=EC=8B=9C=20=EC=82=AC=EC=9A=A9=EB=90=98=EB=8A=94=20=EC=BD=94?=
 =?UTF-8?q?=EB=93=9C=20=EC=B6=94=EA=B0=80=20=EB=B0=8F=20=EB=AC=B8=EA=B5=AC?=
 =?UTF-8?q?=EA=B0=80=20=EC=98=A4=EB=A5=98=EC=9D=B8=EA=B1=B8=20=EC=97=90?=
 =?UTF-8?q?=EB=9F=AC=EB=A1=9C=20=EC=88=98=EC=A0=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/core/status.py | 36 +++++++++++++++++++-----------------
 1 file changed, 19 insertions(+), 17 deletions(-)

diff --git a/app/core/status.py b/app/core/status.py
index d20e742..d4359e9 100644
--- a/app/core/status.py
+++ b/app/core/status.py
@@ -20,6 +20,7 @@ class CommonCode(Enum):
     """ DRIVER, DB 성공 코드 - 21xx """
     SUCCESS_DRIVER_INFO = (status.HTTP_200_OK, "2100", "드라이버 정보 조회를 성공하였습니다.")
     SUCCESS_USER_DB_CONNECT_TEST = (status.HTTP_200_OK, "2101", "테스트 연결을 성공하였습니다.")
+    SUCCESS_SAVE_DB_PROFILE = (status.HTTP_200_OK, "2102", "DB 연결 정보를 저장하였습니다.")
 
     """ KEY 성공 코드 - 22xx """
 
@@ -31,19 +32,19 @@ class CommonCode(Enum):
     """ SQL 성공 코드 - 25xx """
 
     # =======================================
-    #    클라이언트 오류 (Client Error) - 4xxx
+    #    클라이언트 에러 (Client Error) - 4xxx
     # =======================================
-    """ 기본 클라이언트 오류 코드 - 40xx """
+    """ 기본 클라이언트 에러 코드 - 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_PARAMETER = (status.HTTP_422_UNPROCESSABLE_ENTITY, "4003", "필수 값이 누락되었습니다.")
 
-    """ DRIVER, DB 클라이언트 오류 코드 - 41xx """
+    """ DRIVER, DB 클라이언트 에러 코드 - 41xx """
     INVALID_DB_DRIVER = (status.HTTP_409_CONFLICT, "4100", "지원하지 않는 데이터베이스입니다.")
     NO_DB_DRIVER = (status.HTTP_400_BAD_REQUEST, "4101", "데이터베이스는 필수 값입니다.")
 
-    """ KEY 클라이언트 오류 코드 - 42xx """
+    """ KEY 클라이언트 에러 코드 - 42xx """
     INVALID_API_KEY_FORMAT = (status.HTTP_400_BAD_REQUEST, "4200", "API 키의 형식이 올바르지 않습니다.")
     INVALID_API_KEY_PREFIX = (
         status.HTTP_400_BAD_REQUEST,
@@ -51,7 +52,7 @@ class CommonCode(Enum):
         "API 키가 선택한 서비스의 올바른 형식이 아닙니다. (예: OpenAI는 sk-로 시작)",
     )
 
-    """ AI CHAT TAB 클라이언트 오류 코드 - 43xx """
+    """ AI CHAT, DB 클라이언트 에러 코드 - 43xx """
     INVALID_CHAT_TAB_NAME_FORMAT = (status.HTTP_400_BAD_REQUEST, "4300", "채팅 탭 이름의 형식이 올바르지 않습니다.")
     INVALID_CHAT_TAB_NAME_LENGTH = (
         status.HTTP_400_BAD_REQUEST,
@@ -65,15 +66,15 @@ class CommonCode(Enum):
         "허용되지 않는 특수 문자: 큰따옴표(\"), 작은따옴표('), 세미콜론(;), 꺾쇠괄호(<, >)",
     )
 
-    """ ANNOTATION 클라이언트 오류 코드 - 44xx """
+    """ ANNOTATION 클라이언트 에러 코드 - 44xx """
 
-    """ SQL 클라이언트 오류 코드 - 45xx """
+    """ SQL 클라이언트 에러 코드 - 45xx """
 
     # ==================================
-    #    서버 오류 (Server Error) - 5xx
+    #    서버 에러 (Server Error) - 5xx
     # ==================================
-    """ 기본 서버 오류 코드 - 50xx """
-    FAIL = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5000", "서버 처리 중 오류가 발생했습니다.")
+    """ 기본 서버 에러 코드 - 50xx """
+    FAIL = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5000", "서버 처리 중 에러가 발생했습니다.")
     DB_BUSY = (
         status.HTTP_503_SERVICE_UNAVAILABLE,
         "5001",
@@ -82,19 +83,20 @@ class CommonCode(Enum):
     FAIL_TO_VERIFY_CREATION = (
         status.HTTP_500_INTERNAL_SERVER_ERROR,
         "5002",
-        "데이터 생성 후 검증 과정에서 오류가 발생했습니다.",
+        "데이터 생성 후 검증 과정에서 에러가 발생했습니다.",
     )
 
-    """ DRIVER, DB 서버 오류 코드 - 51xx """
-    FAIL_CONNECT_DB = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5100", "디비 연결 중 오류가 발생했습니다.")
+    """ DRIVER, DB 서버 에러 코드 - 51xx """
+    FAIL_CONNECT_DB = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5100", "디비 연결 중 에러가 발생했습니다.")
+    FAIL_SAVE_PROFILE = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5100", "디비 정보 저장 중 에러가 발생했습니다.")
 
-    """ KEY 서버 오류 코드 - 52xx """
+    """ KEY 서버 에러 코드 - 52xx """
 
-    """ AI CHAT, DB 서버 오류 코드 - 53xx """
+    """ AI CHAT, DB 서버 에러 코드 - 53xx """
 
-    """ ANNOTATION 서버 오류 코드 - 54xx """
+    """ ANNOTATION 서버 에러 코드 - 54xx """
 
-    """ SQL 서버 오류 코드 - 55xx """
+    """ SQL 서버 에러 코드 - 55xx """
 
     def __init__(self, http_status: int, code: str, message: str):
         """Enum 멤버가 생성될 때 각 값을 속성으로 할당합니다."""

From ac0445a4c5d867be4644067dd92227b70589afe6 Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sat, 9 Aug 2025 00:14:10 +0900
Subject: [PATCH 03/25] =?UTF-8?q?refactor:=20=EC=BB=AC=EB=9F=BC=20?=
 =?UTF-8?q?=EC=84=A4=EC=A0=95=20=EB=B3=80=EA=B2=BD=20=EB=B0=8F=20=EC=B6=94?=
 =?UTF-8?q?=EA=B0=80,=20=EC=A0=9C=EA=B1=B0=20=EC=8B=9C=20=EC=97=85?=
 =?UTF-8?q?=EB=8D=B0=EC=9D=B4=ED=8A=B8=20=EB=90=98=EB=8F=84=EB=A1=9D=20?=
 =?UTF-8?q?=EC=88=98=EC=A0=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/db/init_db.py | 235 +++++++++++++++++++++++++---------------------
 1 file changed, 127 insertions(+), 108 deletions(-)

diff --git a/app/db/init_db.py b/app/db/init_db.py
index 34a8c95..0989e5c 100644
--- a/app/db/init_db.py
+++ b/app/db/init_db.py
@@ -1,152 +1,171 @@
 # db/init_db.py
 import sqlite3
+import logging
 
 from app.core.utils import get_db_path
 
-"""
-데이터베이스에 연결하고, 애플리케이션에 필요한 테이블이 없으면 생성합니다.
-"""
+def _synchronize_table(cursor, table_name: str, target_columns: dict):
+    """
+    테이블 스키마를 확인하고, 코드와 다를 경우 테이블을 재생성하여 동기화합니다.
+    """
+    try:
+        cursor.execute(f"PRAGMA table_info({table_name})")
+        current_schema_rows = cursor.fetchall()
+        current_columns = {row[1]: row[2].upper() for row in current_schema_rows}
 
+        target_schema_simple = {name: definition.split()[0].upper() for name, definition in target_columns.items()}
 
-def initialize_database():
+        if current_columns == target_schema_simple:
+            return
+
+        logging.warning(f"'{table_name}' 테이블의 스키마 변경을 감지했습니다. 마이그레이션을 시작합니다. (데이터 손실 위험)")
+
+        temp_table_name = f"{table_name}_temp_old"
+        cursor.execute(f"ALTER TABLE {table_name} RENAME TO {temp_table_name}")
+
+        columns_with_definitions = ", ".join([f"{name} {definition}" for name, definition in target_columns.items()])
+        cursor.execute(f"CREATE TABLE {table_name} ({columns_with_definitions})")
+
+        cursor.execute(f"PRAGMA table_info({temp_table_name})")
+        temp_columns = {row[1] for row in cursor.fetchall()}
+        common_columns = ", ".join(target_columns.keys() & temp_columns)
+
+        if common_columns:
+            cursor.execute(f"INSERT INTO {table_name} ({common_columns}) SELECT {common_columns} FROM {temp_table_name}")
+            logging.info(f"'{temp_table_name}'에서 '{table_name}'으로 데이터를 복사했습니다.")
+
+        cursor.execute(f"DROP TABLE {temp_table_name}")
+        logging.info(f"임시 테이블 '{temp_table_name}'을(를) 삭제했습니다.")
+
+    except sqlite3.Error as e:
+        logging.error(f"'{table_name}' 테이블 마이그레이션 중 오류 발생: {e}")
+        raise e
 
+
+def initialize_database():
+    """
+    데이터베이스에 연결하고, 테이블 스키마를 최신 상태로 동기화합니다.
+    """
     db_path = get_db_path()
     conn = None
     try:
         conn = sqlite3.connect(db_path)
+        conn.execute("BEGIN")
         cursor = conn.cursor()
-        # db_profile 테이블 생성
-        cursor.execute(
-            """
-            CREATE TABLE IF NOT EXISTS db_profile (
-                id VARCHAR(64) PRIMARY KEY NOT NULL,
-                type VARCHAR(32) NOT NULL,
-                host VARCHAR(255) NOT NULL,
-                port INTEGER NOT NULL,
-                name VARCHAR(64),
-                username VARCHAR(128) NOT NULL,
-                password VARCHAR(128) NOT NULL,
-                created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
-                updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
-            );
-        """
-        )
-        # db_profile 테이블의 updated_at을 자동으로 업데이트하는 트리거
+
+        # --- db_profile 테이블 처리 ---
+        db_profile_cols = {
+            "id": "VARCHAR(64) PRIMARY KEY NOT NULL",
+            "type": "VARCHAR(32) NOT NULL",
+            "host": "VARCHAR(255)",
+            "port": "INTEGER",
+            "name": "VARCHAR(64)",
+            "username": "VARCHAR(128)",
+            "password": "VARCHAR(128)",
+            "view_name": "VARCHAR(64)",
+            "created_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP",
+            "updated_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP"
+        }
+        cursor.execute(f"CREATE TABLE IF NOT EXISTS db_profile ({', '.join([f'{k} {v}' for k, v in db_profile_cols.items()])})")
+        _synchronize_table(cursor, "db_profile", db_profile_cols)
+
         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(
+            BEFORE UPDATE ON db_profile FOR EACH ROW
+            BEGIN UPDATE db_profile SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; END;
             """
-            CREATE TABLE IF NOT EXISTS ai_credential (
-                id VARCHAR(64) PRIMARY KEY 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
-            );
-        """
         )
-        # ai_credential 테이블의 updated_at을 자동으로 업데이트하는 트리거
+
+        # --- ai_credential 테이블 처리 ---
+        ai_credential_cols = {
+            "id": "VARCHAR(64) PRIMARY KEY 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"
+        }
+        cursor.execute(f"CREATE TABLE IF NOT EXISTS ai_credential ({', '.join([f'{k} {v}' for k, v in ai_credential_cols.items()])})")
+        _synchronize_table(cursor, "ai_credential", ai_credential_cols)
+
         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(
+            BEFORE UPDATE ON ai_credential FOR EACH ROW
+            BEGIN UPDATE ai_credential SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; END;
             """
-            CREATE TABLE IF NOT EXISTS chat_tab (
-                id VARCHAR(64) PRIMARY KEY NOT NULL,
-                name VARCHAR(128),
-                created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
-                updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
-            );
-        """
         )
-        # chat_tab 테이블의 updated_at을 자동으로 업데이트하는 트리거
+
+        # --- chat_tab 테이블 처리 ---
+        chat_tab_cols = {
+            "id": "VARCHAR(64) PRIMARY KEY NOT NULL",
+            "name": "VARCHAR(128)",
+            "created_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP",
+            "updated_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP"
+        }
+        cursor.execute(f"CREATE TABLE IF NOT EXISTS chat_tab ({', '.join([f'{k} {v}' for k, v in chat_tab_cols.items()])})")
+        _synchronize_table(cursor, "chat_tab", chat_tab_cols)
         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(
+            BEFORE UPDATE ON chat_tab FOR EACH ROW
+            BEGIN UPDATE chat_tab SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; END;
             """
-            CREATE TABLE IF NOT EXISTS chat_message (
-                id VARCHAR(64) PRIMARY KEY NOT NULL,
-                chat_tab_id VARCHAR(64) NOT NULL,
-                sender VARCHAR(1) NOT NULL,
-                message TEXT NOT NULL,
-                created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
-                updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
-                FOREIGN KEY (chat_tab_id) REFERENCES chat_tab(id)
-            );
-        """
         )
-        # chat_message 테이블의 updated_at을 자동으로 업데이트하는 트리거
+
+        # --- chat_message 테이블 처리 ---
+        chat_message_cols = {
+            "id": "VARCHAR(64) PRIMARY KEY NOT NULL",
+            "chat_tab_id": "VARCHAR(64) NOT NULL",
+            "sender": "VARCHAR(1) NOT NULL",
+            "message": "TEXT NOT NULL",
+            "created_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP",
+            "updated_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP",
+            "FOREIGN KEY (chat_tab_id)": "REFERENCES chat_tab(id)"
+        }
+        create_chat_message_sql = ", ".join([f"{k} {v}" for k, v in chat_message_cols.items() if not k.startswith("FOREIGN KEY")])
+        create_chat_message_sql += f", FOREIGN KEY (chat_tab_id) REFERENCES chat_tab(id)"
+        cursor.execute(f"CREATE TABLE IF NOT EXISTS chat_message ({create_chat_message_sql})")
+        _synchronize_table(cursor, "chat_message", {k: v for k, v in chat_message_cols.items() if not k.startswith("FOREIGN KEY")})
+
         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(
+            BEFORE UPDATE ON chat_message FOR EACH ROW
+            BEGIN UPDATE chat_message SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; END;
             """
-            CREATE TABLE IF NOT EXISTS query_history (
-                id VARCHAR(64) PRIMARY KEY NOT NULL,
-                chat_message_id VARCHAR(64) NOT NULL,
-                query_text TEXT NOT NULL,
-                is_success VARCHAR(1) NOT NULL,
-                error_message TEXT NOT NULL,
-                created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
-                updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
-                FOREIGN KEY (chat_message_id) REFERENCES chat_message(id)
-            );
-        """
         )
-        # query_history 테이블의 updated_at을 자동으로 업데이트하는 트리거
+
+        # --- query_history 테이블 처리 ---
+        query_history_cols = {
+            "id": "VARCHAR(64) PRIMARY KEY NOT NULL",
+            "chat_message_id": "VARCHAR(64) NOT NULL",
+            "query_text": "TEXT NOT NULL",
+            "is_success": "VARCHAR(1) NOT NULL",
+            "error_message": "TEXT NOT NULL",
+            "created_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP",
+            "updated_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP",
+            "FOREIGN KEY (chat_message_id)": "REFERENCES chat_message(id)"
+        }
+        create_query_history_sql = ", ".join([f"{k} {v}" for k, v in query_history_cols.items() if not k.startswith("FOREIGN KEY")])
+        create_query_history_sql += f", FOREIGN KEY (chat_message_id) REFERENCES chat_message(id)"
+        cursor.execute(f"CREATE TABLE IF NOT EXISTS query_history ({create_query_history_sql})")
+        _synchronize_table(cursor, "query_history", {k: v for k, v in query_history_cols.items() if not k.startswith("FOREIGN KEY")})
+
         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;
-        """
+            BEFORE UPDATE ON query_history FOR EACH ROW
+            BEGIN UPDATE query_history SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; END;
+            """
         )
 
         conn.commit()
 
     except sqlite3.Error as e:
-        print(f"데이터베이스 초기화 중 오류 발생: {e}")
+        logging.error(f"데이터베이스 초기화 중 오류 발생: {e}. 변경 사항을 롤백합니다.")
+        if conn:
+            conn.rollback()
     finally:
         if conn:
             conn.close()

From b23b8c1f48e8d8a8cdf8bf26fc0cdaaa6e68e054 Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sat, 9 Aug 2025 00:15:18 +0900
Subject: [PATCH 04/25] =?UTF-8?q?refactor:=20result=EB=A5=BC=20=EA=B3=B5?=
 =?UTF-8?q?=ED=86=B5=EC=9C=BC=EB=A1=9C=20=EC=82=AC=EC=9A=A9=20=ED=95=98?=
 =?UTF-8?q?=EB=8F=84=EB=A1=9D=20=EC=88=98=EC=A0=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/schemas/user_db/connect_test_result_model.py | 10 ----------
 app/schemas/user_db/result_model.py              | 15 +++++++++++++++
 2 files changed, 15 insertions(+), 10 deletions(-)
 delete mode 100644 app/schemas/user_db/connect_test_result_model.py
 create mode 100644 app/schemas/user_db/result_model.py

diff --git a/app/schemas/user_db/connect_test_result_model.py b/app/schemas/user_db/connect_test_result_model.py
deleted file mode 100644
index 5f68f38..0000000
--- a/app/schemas/user_db/connect_test_result_model.py
+++ /dev/null
@@ -1,10 +0,0 @@
-# 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="결과 코드")
diff --git a/app/schemas/user_db/result_model.py b/app/schemas/user_db/result_model.py
new file mode 100644
index 0000000..edc3d47
--- /dev/null
+++ b/app/schemas/user_db/result_model.py
@@ -0,0 +1,15 @@
+# app/schemas/user_db/result_model.py
+
+from pydantic import BaseModel, Field
+
+from app.core.status import CommonCode
+
+# 기본 반환 모델
+class BasicResult(BaseModel):
+    is_successful: bool = Field(..., description="성공 여부")
+    code: CommonCode = Field(None, description="결과 코드")
+
+# 디비 정보 저장
+class SaveProfileResult(BasicResult):
+    """DB 조회 결과를 위한 확장 모델"""
+    name: str = Field(..., description="저장된 디비명")

From c6ad827042cd47bd708962bb992d2e6ada91f06b Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sat, 9 Aug 2025 00:16:09 +0900
Subject: [PATCH 05/25] =?UTF-8?q?refactor:=20db=5Fprofile=EC=9D=84=20?=
 =?UTF-8?q?=EA=B3=B5=ED=86=B5=EC=9C=BC=EB=A1=9C=20=EC=82=AC=EC=9A=A9?=
 =?UTF-8?q?=ED=95=98=EB=8F=84=EB=A1=9D=20=EC=88=98=EC=A0=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/schemas/user_db/db_profile_model.py | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/app/schemas/user_db/db_profile_model.py b/app/schemas/user_db/db_profile_model.py
index 616156c..14e1341 100644
--- a/app/schemas/user_db/db_profile_model.py
+++ b/app/schemas/user_db/db_profile_model.py
@@ -10,14 +10,13 @@
 
 
 # 사용자가 직접 입력해야 하는 정보만 포함합니다.
-class DBProfileCreate(BaseModel):
+class DBProfileInfo(BaseModel):
     type: str = Field(..., description="DB 종류")
     host: str | None = Field(None, description="호스트 주소")
     port: int | None = Field(None, description="포트 번호")
+    name: str | None = Field(None, description="연결할 데이터베이스명")
     username: str | None = Field(None, description="사용자 이름")
     password: str | None = Field(None, description="비밀번호")
-    name: str | None = Field(None, description="데이터베이스 이름")
-    driver: str | None = Field(None, description="드라이버 이름")
 
     def validate_required_fields(self) -> None:
         """DB 종류별 필수 필드 유효성 검사"""
@@ -54,6 +53,9 @@ def _is_empty(value: Any | None) -> bool:
             return True
         return False
 
+class SaveDBProfile(DBProfileInfo):
+    id: str | None = Field(None, description="DB Key 값")
+    view_name: str | None = Field(None, description="DB 노출명")
 
 # DB에서 조회되는 모든 정보를 담는 클래스입니다.
 class DBProfile(BaseModel):

From 5b457ad4bdb0d9d98582b92b4bea4545bfd7694d Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sat, 9 Aug 2025 00:25:35 +0900
Subject: [PATCH 06/25] =?UTF-8?q?style:=20name=EC=9D=84=20view=5Fname?=
 =?UTF-8?q?=EC=9C=BC=EB=A1=9C=20=EB=B3=80=EC=88=98=20=EB=AA=85=EC=B9=AD=20?=
 =?UTF-8?q?=EC=88=98=EC=A0=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/schemas/user_db/result_model.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/app/schemas/user_db/result_model.py b/app/schemas/user_db/result_model.py
index edc3d47..e3461d7 100644
--- a/app/schemas/user_db/result_model.py
+++ b/app/schemas/user_db/result_model.py
@@ -12,4 +12,4 @@ class BasicResult(BaseModel):
 # 디비 정보 저장
 class SaveProfileResult(BasicResult):
     """DB 조회 결과를 위한 확장 모델"""
-    name: str = Field(..., description="저장된 디비명")
+    view_name: str = Field(..., description="저장된 디비명")

From be92d224b9cc01764e03c33dc75eb13f49ffee6e Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sat, 9 Aug 2025 00:26:12 +0900
Subject: [PATCH 07/25] =?UTF-8?q?feat:=20db=20profile=20=EC=A0=80=EC=9E=A5?=
 =?UTF-8?q?=ED=95=98=EB=8A=94=20=EB=B6=80=EB=B6=84=20=EC=B6=94=EA=B0=80?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/api/user_db_api.py               | 24 ++++++++++--
 app/repository/user_db_repository.py | 55 +++++++++++++++++++++++++---
 app/services/user_db_service.py      | 32 ++++++++++++----
 3 files changed, 95 insertions(+), 16 deletions(-)

diff --git a/app/api/user_db_api.py b/app/api/user_db_api.py
index f8fec19..37968f3 100644
--- a/app/api/user_db_api.py
+++ b/app/api/user_db_api.py
@@ -4,8 +4,9 @@
 
 from app.core.exceptions import APIException
 from app.core.response import ResponseMessage
-from app.schemas.user_db.db_profile_model import DBProfileCreate
+from app.schemas.user_db.db_profile_model import DBProfileInfo, SaveDBProfile
 from app.services.user_db_service import UserDbService, user_db_service
+from app.schemas.user_db.result_model import SaveProfileResult
 
 user_db_service_dependency = Depends(lambda: user_db_service)
 
@@ -18,13 +19,30 @@
     summary="DB 연결 테스트",
 )
 def connection_test(
-    db_info: DBProfileCreate,
+    db_info: DBProfileInfo,
     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)
+
+@router.post(
+    "/save/profile",
+    response_model=ResponseMessage[str],
+    summary="DB 프로필 저장",
+)
+def save_profile(
+    save_db_info: SaveDBProfile,
+    service: UserDbService = user_db_service_dependency,
+) -> ResponseMessage[str]:
+    """DB 연결 정보를 저장합니다."""
+    save_db_info.validate_required_fields()
+    result = service.save_profile(save_db_info)
+
+    if not result.is_successful:
+        raise APIException(result.code)
+    return ResponseMessage.success(value=result.view_name, code=result.code)
diff --git a/app/repository/user_db_repository.py b/app/repository/user_db_repository.py
index d77d789..7ec4fc3 100644
--- a/app/repository/user_db_repository.py
+++ b/app/repository/user_db_repository.py
@@ -1,13 +1,19 @@
 from typing import Any
 
 import oracledb
+import sqlite3
 
 from app.core.status import CommonCode
-from app.schemas.user_db.connect_test_result_model import TestConnectionResult
-
+from app.schemas.user_db.result_model import BasicResult, SaveProfileResult
+from app.schemas.user_db.db_profile_model import SaveDBProfile
+from app.core.utils import get_db_path
 
 class UserDbRepository:
-    def test_db_connection(self, driver_module: Any, **kwargs: Any) -> TestConnectionResult:
+    def connection_test(
+        self,
+        driver_module: Any,
+        **kwargs: Any
+    ) -> BasicResult:
         """
         DB 드라이버와 연결에 필요한 매개변수들을 받아 연결을 테스트합니다.
         """
@@ -27,13 +33,50 @@ def test_db_connection(self, driver_module: Any, **kwargs: Any) -> TestConnectio
             else:
                 connection = driver_module.connect(**kwargs)
 
-            return TestConnectionResult(is_successful=True, code=CommonCode.SUCCESS_USER_DB_CONNECT_TEST)
+            return BasicResult(is_successful=True, code=CommonCode.SUCCESS_USER_DB_CONNECT_TEST)
+        except (AttributeError, driver_module.OperationalError, driver_module.DatabaseError) as e:
+            return BasicResult(is_successful=False, code=CommonCode.FAIL_CONNECT_DB)
+        except Exception as e:
+            return BasicResult(is_successful=False, code=CommonCode.FAIL)
+        finally:
+            if connection:
+                connection.close()
 
+    def save_profile(
+        self,
+        save_db_info: SaveDBProfile
+    ) -> SaveProfileResult:
+        """
+        DB 드라이버와 연결에 필요한 매개변수들을 받아 연결을 테스트합니다.
+        """
+        db_path = get_db_path()
+        connection = None
+        try:
+            connection = sqlite3.connect(db_path)
+            cursor = connection.cursor()
+            profile_dict = save_db_info.model_dump()
+
+            columns_to_insert = {
+                key: value for key, value in profile_dict.items() if value is not None
+            }
+
+            columns = ", ".join(columns_to_insert.keys())
+            placeholders = ", ".join(["?"] * len(columns_to_insert))
+
+            sql = f"INSERT INTO db_profile ({columns}) VALUES ({placeholders})"
+            data_to_insert = tuple(columns_to_insert.values())
+
+            cursor.execute(sql, data_to_insert)
+            connection.commit()
+            name = save_db_info.view_name if save_db_info.view_name else save_db_info.type
+
+            return SaveProfileResult(is_successful=True, code=CommonCode.SUCCESS_SAVE_DB_PROFILE, view_name=name)
+        except sqlite3.Error:
+            return SaveProfileResult(is_successful=False, code=CommonCode.FAIL_SAVE_PROFILE)
         except Exception:
-            return TestConnectionResult(is_successful=False, code=CommonCode.FAIL_CONNECT_DB)
+            return SaveProfileResult(is_successful=False, code=CommonCode.FAIL_SAVE_PROFILE)
         finally:
             if connection:
                 connection.close()
 
-
 user_db_repository = UserDbRepository()
diff --git a/app/services/user_db_service.py b/app/services/user_db_service.py
index 50b5b9f..4beb961 100644
--- a/app/services/user_db_service.py
+++ b/app/services/user_db_service.py
@@ -10,24 +10,42 @@
 from app.core.exceptions import APIException
 from app.core.status import CommonCode
 from app.repository.user_db_repository import UserDbRepository, user_db_repository
-from app.schemas.user_db.connect_test_result_model import TestConnectionResult
-from app.schemas.user_db.db_profile_model import DBProfileCreate
+from app.schemas.user_db.result_model import BasicResult, SaveProfileResult
+from app.schemas.user_db.db_profile_model import DBProfileInfo, SaveDBProfile
+from app.core.utils import generate_prefixed_uuid
+from app.core.enum.db_key_prefix_name import DBSaveIdEnum
 
 user_db_repository_dependency = Depends(lambda: user_db_repository)
 
 
 class UserDbService:
     def connection_test(
-        self, db_info: DBProfileCreate, repository: UserDbRepository = user_db_repository
-    ) -> TestConnectionResult:
+        self,
+        db_info: DBProfileInfo,
+        repository: UserDbRepository = user_db_repository
+    ) -> BasicResult:
         """
         DB 연결 정보를 받아 연결 테스트를 수행하고 결과를 객체로 반환합니다.
         """
         try:
             driver_module = self._get_driver_module(db_info.type)
             connect_kwargs = self._prepare_connection_args(db_info)
-            return repository.test_db_connection(driver_module, **connect_kwargs)
-        except (ValueError, ImportError) as e:
+            return repository.connection_test(driver_module, **connect_kwargs)
+        except Exception as e:
+            raise APIException(CommonCode.FAIL) from e
+
+    def save_profile(
+        self,
+        save_db_info: SaveDBProfile,
+        repository: UserDbRepository = user_db_repository
+    ) -> SaveProfileResult:
+        """
+        DB 연결 정보를 저장 후 결과를 객체로 반환합니다.
+        """
+        save_db_info.id = generate_prefixed_uuid(DBSaveIdEnum.user_db.value)
+        try:
+            return repository.save_profile(save_db_info)
+        except Exception as e:
             raise APIException(CommonCode.FAIL) from e
 
     def _get_driver_module(self, db_type: str):
@@ -39,7 +57,7 @@ def _get_driver_module(self, db_type: str):
             return sqlite3
         return importlib.import_module(driver_name)
 
-    def _prepare_connection_args(self, db_info: DBProfileCreate) -> dict[str, Any]:
+    def _prepare_connection_args(self, db_info: DBProfileInfo) -> dict[str, Any]:
         """
         DB 타입에 따라 연결에 필요한 매개변수를 딕셔너리로 구성합니다.
         """

From ad016e19c03bee73a935183bf09dd21d0522a089 Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sat, 9 Aug 2025 01:00:32 +0900
Subject: [PATCH 08/25] =?UTF-8?q?feat:=20=EC=A0=84=EC=B2=B4=20=EB=B0=A9?=
 =?UTF-8?q?=EC=83=9D=ED=95=98=EB=8A=94=20=EB=A1=9C=EA=B7=B8=EB=A5=BC=20?=
 =?UTF-8?q?=EC=B0=8D=EC=96=B4=EC=A3=BC=EB=8A=94=20logging=20=EC=B6=94?=
 =?UTF-8?q?=EA=B0=80?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/core/all_logging.py | 33 +++++++++++++++++++++++++++++++++
 app/main.py             |  6 ++++++
 2 files changed, 39 insertions(+)
 create mode 100644 app/core/all_logging.py

diff --git a/app/core/all_logging.py b/app/core/all_logging.py
new file mode 100644
index 0000000..c31d4eb
--- /dev/null
+++ b/app/core/all_logging.py
@@ -0,0 +1,33 @@
+# app/core/all_logging.py
+
+import logging
+from fastapi import Request
+
+# 로깅 기본 설정 (애플리케이션 시작 시 한 번만 구성)
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(levelname)s - %(message)s", # [수정] 로그 레벨(INFO, ERROR)을 포함
+    datefmt="%Y-%m-%d %H:%M:%S",
+)
+
+
+async def log_requests_middleware(request: Request, call_next):
+    """
+    모든 API 요청과 에러에 대한 로그를 남기는 미들웨어입니다.
+    """
+    endpoint = f"{request.method} {request.url.path}"
+
+    # 일반 요청 로그를 남깁니다.
+    logging.info(f"엔드포인트: {endpoint}")
+
+    try:
+        # 다음 미들웨어 또는 실제 엔드포인트를 호출합니다.
+        response = await call_next(request)
+        return response
+    except Exception as e:
+        # [수정] 에러 발생 시, exc_info=True를 추가하여 전체 트레이스백을 함께 기록합니다.
+        # 메시지 형식도 "ERROR 엔드포인트:"로 변경합니다.
+        logging.error(f"ERROR 엔드포인트: {endpoint}", exc_info=True)
+        # 예외를 다시 발생시켜 FastAPI의 전역 예외 처리기가 최종 응답을 만들도록 합니다.
+        raise e
+
diff --git a/app/main.py b/app/main.py
index 198e5fd..01e9da2 100644
--- a/app/main.py
+++ b/app/main.py
@@ -14,8 +14,14 @@
 )
 from app.db.init_db import initialize_database
 
+from starlette.middleware.base import BaseHTTPMiddleware
+from app.core.all_logging import log_requests_middleware
+
 app = FastAPI()
 
+# 전체 로그 찍는 부분
+app.add_middleware(BaseHTTPMiddleware, dispatch=log_requests_middleware)
+
 # 전역 예외 처리기 등록
 app.add_exception_handler(Exception, generic_exception_handler)
 app.add_exception_handler(APIException, api_exception_handler)

From a24d3da2bb9019fb85df235f7c1c14de4483e296 Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sat, 9 Aug 2025 02:14:28 +0900
Subject: [PATCH 09/25] =?UTF-8?q?feat:=20db=20=EC=A0=84=EC=B2=B4=20?=
 =?UTF-8?q?=EC=A1=B0=ED=9A=8C=20=EC=83=81=ED=83=9C=20=EA=B0=92=20=EC=B6=94?=
 =?UTF-8?q?=EA=B0=80?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/core/status.py | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/app/core/status.py b/app/core/status.py
index d4359e9..415106a 100644
--- a/app/core/status.py
+++ b/app/core/status.py
@@ -20,7 +20,9 @@ class CommonCode(Enum):
     """ DRIVER, DB 성공 코드 - 21xx """
     SUCCESS_DRIVER_INFO = (status.HTTP_200_OK, "2100", "드라이버 정보 조회를 성공하였습니다.")
     SUCCESS_USER_DB_CONNECT_TEST = (status.HTTP_200_OK, "2101", "테스트 연결을 성공하였습니다.")
-    SUCCESS_SAVE_DB_PROFILE = (status.HTTP_200_OK, "2102", "DB 연결 정보를 저장하였습니다.")
+    SUCCESS_FIND_ALL_PROFILE = (status.HTTP_200_OK, "2102", "DB 정보 조회를 성공하였습니다.")
+    SUCCESS_SAVE_DB_PROFILE = (status.HTTP_200_OK, "2132", "DB 연결 정보를 저장하였습니다.")
+
 
     """ KEY 성공 코드 - 22xx """
 
@@ -89,6 +91,7 @@ class CommonCode(Enum):
     """ DRIVER, DB 서버 에러 코드 - 51xx """
     FAIL_CONNECT_DB = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5100", "디비 연결 중 에러가 발생했습니다.")
     FAIL_SAVE_PROFILE = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5100", "디비 정보 저장 중 에러가 발생했습니다.")
+    FAIL_FIND_ALL_PROFILE = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5100", "디비 정보 조회 중 에러가 발생했습니다.")
 
     """ KEY 서버 에러 코드 - 52xx """
 

From 2df2757400bdf39d5aa88d6c73b31213489ee781 Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sat, 9 Aug 2025 02:15:33 +0900
Subject: [PATCH 10/25] =?UTF-8?q?refactor:=20=EC=A1=B0=ED=9A=8C=20?=
 =?UTF-8?q?=EA=B2=B0=EA=B3=BC=20=EB=B6=80=EB=B6=84=20result=20=EB=AA=A8?=
 =?UTF-8?q?=EB=8D=B8=EB=A1=9C=20=EC=9D=B4=EB=8F=99?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/schemas/user_db/db_profile_model.py | 14 --------------
 app/schemas/user_db/result_model.py     | 24 +++++++++++++++++++++++-
 2 files changed, 23 insertions(+), 15 deletions(-)

diff --git a/app/schemas/user_db/db_profile_model.py b/app/schemas/user_db/db_profile_model.py
index 14e1341..c7d9348 100644
--- a/app/schemas/user_db/db_profile_model.py
+++ b/app/schemas/user_db/db_profile_model.py
@@ -56,17 +56,3 @@ def _is_empty(value: Any | None) -> bool:
 class SaveDBProfile(DBProfileInfo):
     id: str | None = Field(None, description="DB Key 값")
     view_name: str | None = Field(None, description="DB 노출명")
-
-# DB에서 조회되는 모든 정보를 담는 클래스입니다.
-class DBProfile(BaseModel):
-    id: str
-    type: str
-    host: str
-    port: int
-    name: str | None
-    username: str
-    created_at: datetime
-    updated_at: datetime
-
-    class Config:
-        from_attributes = True
diff --git a/app/schemas/user_db/result_model.py b/app/schemas/user_db/result_model.py
index e3461d7..642e648 100644
--- a/app/schemas/user_db/result_model.py
+++ b/app/schemas/user_db/result_model.py
@@ -1,6 +1,8 @@
 # app/schemas/user_db/result_model.py
 
 from pydantic import BaseModel, Field
+from datetime import datetime
+from typing import List
 
 from app.core.status import CommonCode
 
@@ -9,7 +11,27 @@ class BasicResult(BaseModel):
     is_successful: bool = Field(..., description="성공 여부")
     code: CommonCode = Field(None, description="결과 코드")
 
-# 디비 정보 저장
+# 디비 정보 후 반환되는 저장 모델
 class SaveProfileResult(BasicResult):
     """DB 조회 결과를 위한 확장 모델"""
     view_name: str = Field(..., description="저장된 디비명")
+
+# DB Profile 조회되는 정보를 담는 모델입니다.
+class DBProfile(BaseModel):
+    id: str
+    type: str
+    host: str
+    port: int
+    name: str | None
+    username: str
+    view_name: str | None
+    created_at: datetime
+    updated_at: datetime
+
+    class Config:
+        from_attributes = True
+
+# DB Profile 전체 조회 결과를 담는 새로운 모델
+class AllDBProfileResult(BasicResult):
+    """DB 프로필 전체 조회 결과를 위한 확장 모델"""
+    profiles: List[DBProfile] = Field([], description="DB 프로필 목록")
\ No newline at end of file

From dd07df67141bb9dfbda71ae68a0594f5933d2b6f Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sat, 9 Aug 2025 02:16:58 +0900
Subject: [PATCH 11/25] =?UTF-8?q?feat:=20db=20=EC=A0=84=EC=B2=B4=20?=
 =?UTF-8?q?=EC=A1=B0=ED=9A=8C=20=EA=B8=B0=EB=8A=A5=20=EA=B5=AC=ED=98=84?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/api/user_db_api.py               | 19 +++++++++++++++---
 app/repository/user_db_repository.py | 29 +++++++++++++++++++++++++++-
 app/services/user_db_service.py      | 18 ++++++++++++++---
 3 files changed, 59 insertions(+), 7 deletions(-)

diff --git a/app/api/user_db_api.py b/app/api/user_db_api.py
index 37968f3..68a1754 100644
--- a/app/api/user_db_api.py
+++ b/app/api/user_db_api.py
@@ -1,12 +1,13 @@
 # app/api/user_db_api.py
 
 from fastapi import APIRouter, Depends
+from typing import List
 
 from app.core.exceptions import APIException
 from app.core.response import ResponseMessage
 from app.schemas.user_db.db_profile_model import DBProfileInfo, SaveDBProfile
 from app.services.user_db_service import UserDbService, user_db_service
-from app.schemas.user_db.result_model import SaveProfileResult
+from app.schemas.user_db.result_model import DBProfile
 
 user_db_service_dependency = Depends(lambda: user_db_service)
 
@@ -22,7 +23,6 @@ def connection_test(
     db_info: DBProfileInfo,
     service: UserDbService = user_db_service_dependency,
 ) -> ResponseMessage[bool]:
-    """DB 연결 정보를 받아 연결 가능 여부를 테스트합니다."""
     db_info.validate_required_fields()
     result = service.connection_test(db_info)
 
@@ -39,10 +39,23 @@ def save_profile(
     save_db_info: SaveDBProfile,
     service: UserDbService = user_db_service_dependency,
 ) -> ResponseMessage[str]:
-    """DB 연결 정보를 저장합니다."""
     save_db_info.validate_required_fields()
     result = service.save_profile(save_db_info)
 
     if not result.is_successful:
         raise APIException(result.code)
     return ResponseMessage.success(value=result.view_name, code=result.code)
+
+@router.get(
+    "/find/all",
+    response_model=ResponseMessage[List[DBProfile]],
+    summary="DB 프로필 전체 조회",
+)
+def find_all_profile(
+    service: UserDbService = user_db_service_dependency,
+) -> ResponseMessage[List[DBProfile]]:
+    result = service.find_all_profile()
+
+    if not result.is_successful:
+        raise APIException(result.code)
+    return ResponseMessage.success(value=result.profiles, code=result.code)
diff --git a/app/repository/user_db_repository.py b/app/repository/user_db_repository.py
index 7ec4fc3..9ed14c7 100644
--- a/app/repository/user_db_repository.py
+++ b/app/repository/user_db_repository.py
@@ -4,7 +4,7 @@
 import sqlite3
 
 from app.core.status import CommonCode
-from app.schemas.user_db.result_model import BasicResult, SaveProfileResult
+from app.schemas.user_db.result_model import BasicResult, SaveProfileResult, AllDBProfileResult, DBProfile
 from app.schemas.user_db.db_profile_model import SaveDBProfile
 from app.core.utils import get_db_path
 
@@ -79,4 +79,31 @@ def save_profile(
             if connection:
                 connection.close()
 
+    def find_all_profile(
+        self
+    ) -> AllDBProfileResult:
+        """
+        모든 DB 연결 정보를 조회합니다.
+        """
+        db_path = get_db_path()
+        connection = None
+        try:
+            connection = sqlite3.connect(db_path)
+            connection.row_factory = sqlite3.Row
+            cursor = connection.cursor()
+
+            sql = "SELECT * FROM db_profile"
+            cursor.execute(sql)
+            rows = cursor.fetchall()
+            profiles = [DBProfile(**row) for row in rows]
+
+            return AllDBProfileResult(is_successful=True, code=CommonCode.SUCCESS_FIND_ALL_PROFILE, profiles=profiles)
+        except sqlite3.Error:
+            return AllDBProfileResult(is_successful=False, code=CommonCode.FAIL_FIND_ALL_PROFILE)
+        except Exception:
+            return AllDBProfileResult(is_successful=False, code=CommonCode.FAIL_FIND_ALL_PROFILE)
+        finally:
+            if connection:
+                connection.close()
+
 user_db_repository = UserDbRepository()
diff --git a/app/services/user_db_service.py b/app/services/user_db_service.py
index 4beb961..67d05af 100644
--- a/app/services/user_db_service.py
+++ b/app/services/user_db_service.py
@@ -10,7 +10,7 @@
 from app.core.exceptions import APIException
 from app.core.status import CommonCode
 from app.repository.user_db_repository import UserDbRepository, user_db_repository
-from app.schemas.user_db.result_model import BasicResult, SaveProfileResult
+from app.schemas.user_db.result_model import BasicResult, SaveProfileResult, AllDBProfileResult
 from app.schemas.user_db.db_profile_model import DBProfileInfo, SaveDBProfile
 from app.core.utils import generate_prefixed_uuid
 from app.core.enum.db_key_prefix_name import DBSaveIdEnum
@@ -25,7 +25,7 @@ def connection_test(
         repository: UserDbRepository = user_db_repository
     ) -> BasicResult:
         """
-        DB 연결 정보를 받아 연결 테스트를 수행하고 결과를 객체로 반환합니다.
+        DB 연결 정보를 받아 연결 테스트를 수행 후 결과를 반환합니다.
         """
         try:
             driver_module = self._get_driver_module(db_info.type)
@@ -40,7 +40,7 @@ def save_profile(
         repository: UserDbRepository = user_db_repository
     ) -> SaveProfileResult:
         """
-        DB 연결 정보를 저장 후 결과를 객체로 반환합니다.
+        DB 연결 정보를 저장 후 결과를 반환합니다.
         """
         save_db_info.id = generate_prefixed_uuid(DBSaveIdEnum.user_db.value)
         try:
@@ -48,6 +48,18 @@ def save_profile(
         except Exception as e:
             raise APIException(CommonCode.FAIL) from e
 
+    def find_all_profile(
+        self,
+        repository: UserDbRepository = user_db_repository
+    ) -> AllDBProfileResult:
+        """
+        모든 DB 연결 정보를 반환합니다.
+        """
+        try:
+            return repository.find_all_profile()
+        except Exception as e:
+            raise APIException(CommonCode.FAIL) from e
+
     def _get_driver_module(self, db_type: str):
         """
         DB 타입에 따라 동적으로 드라이버 모듈을 로드합니다.

From 05b5030164a468423fd3bcaa1f18e9c6b61c44c5 Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sun, 10 Aug 2025 17:52:21 +0900
Subject: [PATCH 12/25] =?UTF-8?q?feat:=20=EB=B0=98=ED=99=98=20=EC=83=81?=
 =?UTF-8?q?=ED=83=9C=20=EA=B0=92=20=EC=B6=94=EA=B0=80?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/core/status.py | 14 ++++++++++----
 1 file changed, 10 insertions(+), 4 deletions(-)

diff --git a/app/core/status.py b/app/core/status.py
index 415106a..2f3f383 100644
--- a/app/core/status.py
+++ b/app/core/status.py
@@ -20,8 +20,11 @@ class CommonCode(Enum):
     """ DRIVER, DB 성공 코드 - 21xx """
     SUCCESS_DRIVER_INFO = (status.HTTP_200_OK, "2100", "드라이버 정보 조회를 성공하였습니다.")
     SUCCESS_USER_DB_CONNECT_TEST = (status.HTTP_200_OK, "2101", "테스트 연결을 성공하였습니다.")
-    SUCCESS_FIND_ALL_PROFILE = (status.HTTP_200_OK, "2102", "DB 정보 조회를 성공하였습니다.")
-    SUCCESS_SAVE_DB_PROFILE = (status.HTTP_200_OK, "2132", "DB 연결 정보를 저장하였습니다.")
+    SUCCESS_FIND_PROFILE = (status.HTTP_200_OK, "2102", "디비 정보 조회를 성공하였습니다.")
+    SUCCESS_FIND_SCHEMAS = (status.HTTP_200_OK, "2103", "디비 스키마 정보 조회를 성공하였습니다.")
+    SUCCESS_FIND_TABLES = (status.HTTP_200_OK, "2104", "디비 테이블 정보 조회를 성공하였습니다.")
+    SUCCESS_FIND_COLUMNS = (status.HTTP_200_OK, "2105", "디비 컬럼 정보 조회를 성공하였습니다.")
+    SUCCESS_SAVE_DB_PROFILE = (status.HTTP_200_OK, "2130", "디비 연결 정보를 저장하였습니다.")
 
 
     """ KEY 성공 코드 - 22xx """
@@ -90,8 +93,11 @@ class CommonCode(Enum):
 
     """ DRIVER, DB 서버 에러 코드 - 51xx """
     FAIL_CONNECT_DB = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5100", "디비 연결 중 에러가 발생했습니다.")
-    FAIL_SAVE_PROFILE = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5100", "디비 정보 저장 중 에러가 발생했습니다.")
-    FAIL_FIND_ALL_PROFILE = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5100", "디비 정보 조회 중 에러가 발생했습니다.")
+    FAIL_SAVE_PROFILE = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5101", "디비 정보 저장 중 에러가 발생했습니다.")
+    FAIL_FIND_PROFILE = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5102", "디비 정보 조회 중 에러가 발생했습니다.")
+    FAIL_FIND_SCHEMAS = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5103", "디비 스키마 정보 조회 중 에러가 발생했습니다.")
+    FAIL_FIND_TABLES = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5104", "디비 테이블 정보 조회 중 에러가 발생했습니다.")
+    FAIL_FIND_COLUMNS = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5105", "디비 컬럼 정보 조회 중 에러가 발생했습니다.")
 
     """ KEY 서버 에러 코드 - 52xx """
 

From 846a7e7827d027aabbb6756d2d4552e6ede56ea2 Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sun, 10 Aug 2025 17:52:45 +0900
Subject: [PATCH 13/25] =?UTF-8?q?feat:=20=EB=B0=98=ED=99=98=20=ED=83=80?=
 =?UTF-8?q?=EC=9E=85=20=EC=B6=94=EA=B0=80?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/schemas/user_db/db_profile_model.py |  7 ++++-
 app/schemas/user_db/result_model.py     | 38 +++++++++++++++++++++----
 2 files changed, 39 insertions(+), 6 deletions(-)

diff --git a/app/schemas/user_db/db_profile_model.py b/app/schemas/user_db/db_profile_model.py
index c7d9348..ad312b1 100644
--- a/app/schemas/user_db/db_profile_model.py
+++ b/app/schemas/user_db/db_profile_model.py
@@ -17,7 +17,6 @@ class DBProfileInfo(BaseModel):
     name: str | None = Field(None, description="연결할 데이터베이스명")
     username: str | None = Field(None, description="사용자 이름")
     password: str | None = Field(None, description="비밀번호")
-
     def validate_required_fields(self) -> None:
         """DB 종류별 필수 필드 유효성 검사"""
         required_fields_by_type = {
@@ -56,3 +55,9 @@ def _is_empty(value: Any | None) -> bool:
 class SaveDBProfile(DBProfileInfo):
     id: str | None = Field(None, description="DB Key 값")
     view_name: str | None = Field(None, description="DB 노출명")
+
+class AllDBProfileInfo(DBProfileInfo):
+    id: str | None = Field(..., description="DB Key 값")
+    view_name: str | None = Field(None, description="DB 노출명")
+    created_at: datetime = Field(..., description="profile 저장일")
+    updated_at: datetime = Field(..., description="profile 수정일")
diff --git a/app/schemas/user_db/result_model.py b/app/schemas/user_db/result_model.py
index 642e648..7b8f366 100644
--- a/app/schemas/user_db/result_model.py
+++ b/app/schemas/user_db/result_model.py
@@ -2,7 +2,7 @@
 
 from pydantic import BaseModel, Field
 from datetime import datetime
-from typing import List
+from typing import List, Any
 
 from app.core.status import CommonCode
 
@@ -20,10 +20,10 @@ class SaveProfileResult(BasicResult):
 class DBProfile(BaseModel):
     id: str
     type: str
-    host: str
-    port: int
+    host: str | None
+    port: int | None
     name: str | None
-    username: str
+    username: str | None
     view_name: str | None
     created_at: datetime
     updated_at: datetime
@@ -34,4 +34,32 @@ class Config:
 # DB Profile 전체 조회 결과를 담는 새로운 모델
 class AllDBProfileResult(BasicResult):
     """DB 프로필 전체 조회 결과를 위한 확장 모델"""
-    profiles: List[DBProfile] = Field([], description="DB 프로필 목록")
\ No newline at end of file
+    profiles: List[DBProfile] = Field([], description="DB 프로필 목록")
+
+class ColumnInfo(BaseModel):
+    """단일 컬럼의 상세 정보를 담는 모델"""
+    name: str = Field(..., description="컬럼 이름")
+    type: str = Field(..., description="데이터 타입")
+    nullable: bool = Field(..., description="NULL 허용 여부")
+    default: Any | None = Field(None, description="기본값")
+    comment: str | None = Field(None, description="코멘트")
+    is_pk: bool = Field(False, description="기본 키(Primary Key) 여부")
+
+class TableInfo(BaseModel):
+    """단일 테이블의 이름과 컬럼 목록을 담는 모델"""
+    name: str = Field(..., description="테이블 이름")
+    columns: List[ColumnInfo] = Field([], description="컬럼 목록")
+    comment: str | None = Field(None, description="테이블 코멘트")
+
+class SchemaInfoResult(BasicResult):
+    """DB 스키마 상세 정보 조회 결과를 위한 확장 모델"""
+    schema: List[TableInfo] = Field([], description="테이블 및 컬럼 정보 목록")
+
+class SchemaListResult(BasicResult):
+    schemas: List[str] = Field([], description="스키마 이름 목록")
+
+class TableListResult(BasicResult):
+    tables: List[str] = Field([], description="테이블 이름 목록")
+
+class ColumnListResult(BasicResult):
+    columns: List[ColumnInfo] = Field([], description="컬럼 정보 목록")

From e746eed5da3373c2112dd517ff69ecfee065b884 Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sun, 10 Aug 2025 17:53:05 +0900
Subject: [PATCH 14/25] =?UTF-8?q?feat:=20=EC=8A=A4=ED=82=A4=EB=A7=88,=20?=
 =?UTF-8?q?=ED=85=8C=EC=9D=B4=EB=B8=94,=20=EC=BB=AC=EB=9F=BC=20=EC=A1=B0?=
 =?UTF-8?q?=ED=9A=8C=20=EB=A1=9C=EC=A7=81=20=EA=B5=AC=ED=98=84?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/api/user_db_api.py               |  58 ++++++-
 app/repository/user_db_repository.py | 226 ++++++++++++++++++++++++++-
 app/services/user_db_service.py      | 172 +++++++++++++++++++-
 3 files changed, 446 insertions(+), 10 deletions(-)

diff --git a/app/api/user_db_api.py b/app/api/user_db_api.py
index 68a1754..e1bf203 100644
--- a/app/api/user_db_api.py
+++ b/app/api/user_db_api.py
@@ -7,7 +7,7 @@
 from app.core.response import ResponseMessage
 from app.schemas.user_db.db_profile_model import DBProfileInfo, SaveDBProfile
 from app.services.user_db_service import UserDbService, user_db_service
-from app.schemas.user_db.result_model import DBProfile
+from app.schemas.user_db.result_model import DBProfile, TableInfo, ColumnInfo, SchemaListResult, TableListResult, ColumnListResult
 
 user_db_service_dependency = Depends(lambda: user_db_service)
 
@@ -39,6 +39,7 @@ def save_profile(
     save_db_info: SaveDBProfile,
     service: UserDbService = user_db_service_dependency,
 ) -> ResponseMessage[str]:
+
     save_db_info.validate_required_fields()
     result = service.save_profile(save_db_info)
 
@@ -54,8 +55,63 @@ def save_profile(
 def find_all_profile(
     service: UserDbService = user_db_service_dependency,
 ) -> ResponseMessage[List[DBProfile]]:
+
     result = service.find_all_profile()
 
     if not result.is_successful:
         raise APIException(result.code)
     return ResponseMessage.success(value=result.profiles, code=result.code)
+
+@router.get(
+    "/find/schemas/{profile_id}",
+    response_model=ResponseMessage[List[str]],
+    summary="특정 DB의 전체 스키마 조회",
+)
+def find_schemas(
+    profile_id: str,
+    service: UserDbService = user_db_service_dependency
+) -> ResponseMessage[List[str]]:
+
+    db_info = service.find_profile(profile_id)
+    result = service.find_schemas(db_info)
+
+    if not result.is_successful:
+        raise APIException(result.code)
+    return ResponseMessage.success(value=result.schemas, code=result.code)
+
+@router.get(
+    "/find/tables/{profile_id}/{schema_name}",
+    response_model=ResponseMessage[List[str]],
+    summary="특정 스키마의 전체 테이블 조회",
+)
+def find_tables(
+    profile_id: str,
+    schema_name: str,
+    service: UserDbService = user_db_service_dependency
+) -> ResponseMessage[List[str]]:
+
+    db_info = service.find_profile(profile_id)
+    result = service.find_tables(db_info, schema_name)
+
+    if not result.is_successful:
+        raise APIException(result.code)
+    return ResponseMessage.success(value=result.tables, code=result.code)
+
+@router.get(
+    "/find/columns/{profile_id}/{schema_name}/{table_name}",
+    response_model=ResponseMessage[List[ColumnInfo]],
+    summary="특정 테이블의 전체 컬럼 조회",
+)
+def find_columns(
+    profile_id: str,
+    schema_name: str,
+    table_name: str,
+    service: UserDbService = user_db_service_dependency
+) -> ResponseMessage[List[ColumnInfo]]:
+
+    db_info = service.find_profile(profile_id)
+    result = service.find_columns(db_info, schema_name, table_name)
+
+    if not result.is_successful:
+        raise APIException(result.code)
+    return ResponseMessage.success(value=result.columns, code=result.code)
diff --git a/app/repository/user_db_repository.py b/app/repository/user_db_repository.py
index 9ed14c7..2f156c0 100644
--- a/app/repository/user_db_repository.py
+++ b/app/repository/user_db_repository.py
@@ -4,9 +4,22 @@
 import sqlite3
 
 from app.core.status import CommonCode
-from app.schemas.user_db.result_model import BasicResult, SaveProfileResult, AllDBProfileResult, DBProfile
-from app.schemas.user_db.db_profile_model import SaveDBProfile
 from app.core.utils import get_db_path
+from app.core.exceptions import APIException
+from app.schemas.user_db.result_model import (
+    DBProfile,
+    BasicResult,
+    SaveProfileResult,
+    AllDBProfileResult,
+    SchemaListResult,
+    TableListResult,
+    ColumnListResult,
+    ColumnInfo
+)
+from app.schemas.user_db.db_profile_model import (
+    SaveDBProfile,
+    AllDBProfileInfo
+)
 
 class UserDbRepository:
     def connection_test(
@@ -92,18 +105,219 @@ def find_all_profile(
             connection.row_factory = sqlite3.Row
             cursor = connection.cursor()
 
-            sql = "SELECT * FROM db_profile"
+            sql = """
+            SELECT
+                id,
+                type,
+                host,
+                port,
+                name,
+                username,
+                view_name,
+                created_at,
+                updated_at 
+            FROM db_profile
+            """
             cursor.execute(sql)
             rows = cursor.fetchall()
             profiles = [DBProfile(**row) for row in rows]
 
-            return AllDBProfileResult(is_successful=True, code=CommonCode.SUCCESS_FIND_ALL_PROFILE, profiles=profiles)
+            return AllDBProfileResult(is_successful=True, code=CommonCode.SUCCESS_FIND_PROFILE, profiles=profiles)
         except sqlite3.Error:
-            return AllDBProfileResult(is_successful=False, code=CommonCode.FAIL_FIND_ALL_PROFILE)
+            return AllDBProfileResult(is_successful=False, code=CommonCode.FAIL_FIND_PROFILE)
         except Exception:
-            return AllDBProfileResult(is_successful=False, code=CommonCode.FAIL_FIND_ALL_PROFILE)
+            return AllDBProfileResult(is_successful=False, code=CommonCode.FAIL_FIND_PROFILE)
         finally:
             if connection:
                 connection.close()
 
+    def find_profile(
+        self,
+        profile_id
+    ) -> AllDBProfileInfo:
+        """
+        특정 DB 연결 정보를 조회합니다.
+        """
+        db_path = get_db_path()
+        connection = None
+        try:
+            connection = sqlite3.connect(db_path)
+            connection.row_factory = sqlite3.Row
+            cursor = connection.cursor()
+
+            sql = """
+            SELECT
+                id,
+                type,
+                host,
+                port,
+                name,
+                username,
+                password,
+                view_name,
+                created_at,
+                updated_at 
+            FROM db_profile
+            WHERE id = ?
+            """
+            cursor.execute(sql, (profile_id,))
+            row = cursor.fetchone()
+
+            return AllDBProfileInfo(**dict(row))
+        except sqlite3.Error:
+            raise APIException(CommonCode.FAIL_FIND_PROFILE)
+        except Exception:
+            raise APIException(CommonCode.FAIL)
+        finally:
+            if connection:
+                connection.close()
+
+    # ─────────────────────────────
+    # 스키마 조회
+    # ─────────────────────────────
+    def find_schemas(
+            self,
+            driver_module: Any,
+            schema_query: str,
+            **kwargs: Any
+    ) -> SchemaListResult:
+        connection = None
+        try:
+            connection = self._connect(driver_module, **kwargs)
+            cursor = connection.cursor()
+
+            if not schema_query:
+                return SchemaListResult(
+                    is_successful=True,
+                    code=CommonCode.SUCCESS_FIND_SCHEMAS,
+                    schemas=["main"]
+                )
+
+            cursor.execute(schema_query)
+            schemas = [row[0] for row in cursor.fetchall()]
+
+            return SchemaListResult(is_successful=True, code=CommonCode.SUCCESS_FIND_SCHEMAS, schemas=schemas)
+        except Exception:
+            return SchemaListResult(is_successful=False, code=CommonCode.FAIL_FIND_SCHEMAS, schemas=[])
+        finally:
+            if connection:
+                connection.close()
+
+    # ─────────────────────────────
+    # 테이블 조회
+    # ─────────────────────────────
+    def find_tables(
+        self,
+        driver_module: Any,
+        table_query: str,
+        schema_name: str,
+        **kwargs: Any
+    ) -> TableListResult:
+        connection = None
+        try:
+            connection = self._connect(driver_module, **kwargs)
+            cursor = connection.cursor()
+
+            if "%s" in table_query or "?" in table_query:
+                cursor.execute(table_query, (schema_name,))
+            elif ":owner" in table_query:
+                cursor.execute(table_query, {"owner": schema_name})
+            else:
+                cursor.execute(table_query)
+
+            tables = [row[0] for row in cursor.fetchall()]
+
+            return TableListResult(is_successful=True, code=CommonCode.SUCCESS_FIND_TABLES, tables=tables)
+        except Exception:
+            return TableListResult(is_successful=False, code=CommonCode.FAIL_FIND_TABLES, tables=[])
+        finally:
+            if connection:
+                connection.close()
+
+    # ─────────────────────────────
+    # 컬럼 조회
+    # ─────────────────────────────
+    def find_columns(
+        self,
+        driver_module: Any,
+        column_query: str,
+        schema_name: str,
+        db_type,
+        table_name: str,
+        **kwargs: Any
+    ) -> ColumnListResult:
+        connection = None
+        try:
+            connection = self._connect(driver_module, **kwargs)
+            cursor = connection.cursor()
+
+            if db_type == "sqlite":
+                # SQLite는 PRAGMA를 직접 실행
+                pragma_sql = f"PRAGMA table_info('{table_name}')"
+                cursor.execute(pragma_sql)
+                columns_raw = cursor.fetchall()
+                columns = [
+                    ColumnInfo(
+                        name=c[1],
+                        type=c[2],
+                        nullable=(c[3] == 0),  # notnull == 0 means nullable
+                        default=c[4],
+                        comment=None,
+                        is_pk=(c[5] == 1)
+                    )
+                    for c in columns_raw
+                ]
+            else:
+                if "%s" in column_query or "?" in column_query:
+                    cursor.execute(column_query, (schema_name, table_name))
+                elif ":owner" in column_query and ":table" in column_query:
+                    owner_bind = schema_name.upper() if schema_name else schema_name
+                    table_bind = table_name.upper() if table_name else table_name
+                    try:
+                        cursor.execute(column_query, {"owner": owner_bind, "table": table_bind})
+                    except Exception:
+                        # fallback: try positional binds (:1, :2) if named binds fail
+                        try:
+                            pos_query = column_query.replace(":owner", ":1").replace(":table", ":2")
+                            cursor.execute(pos_query, [owner_bind, table_bind])
+                        except Exception:
+                            # re-raise to be handled by outer exception handler
+                            raise
+                else:
+                    cursor.execute(column_query)
+
+                columns = [
+                    ColumnInfo(
+                        name=c[0],
+                        type=c[1],
+                        nullable=(c[2] in ["YES", "Y", True]),
+                        default=c[3],
+                        comment=c[4] if len(c) > 4 else None,
+                        is_pk=(c[5] in ["PRI", True] if len(c) > 5 else False)
+                    )
+                    for c in cursor.fetchall()
+                ]
+
+            return ColumnListResult(is_successful=True, code=CommonCode.SUCCESS_FIND_COLUMNS, columns=columns)
+        except Exception:
+            return ColumnListResult(is_successful=False, code=CommonCode.FAIL_FIND_COLUMNS, columns=[])
+        finally:
+            if connection:
+                connection.close()
+
+    # ─────────────────────────────
+    # DB 연결 메서드
+    # ─────────────────────────────
+    def _connect(self, driver_module: Any, **kwargs):
+        if driver_module is oracledb:
+            if kwargs.get("user", "").lower() == "sys":
+                kwargs["mode"] = oracledb.AUTH_MODE_SYSDBA
+            return driver_module.connect(**kwargs)
+        elif "connection_string" in kwargs:
+            return driver_module.connect(kwargs["connection_string"])
+        elif "db_name" in kwargs:
+            return driver_module.connect(kwargs["db_name"])
+        else:
+            return driver_module.connect(**kwargs)
+
 user_db_repository = UserDbRepository()
diff --git a/app/services/user_db_service.py b/app/services/user_db_service.py
index 67d05af..ead22e7 100644
--- a/app/services/user_db_service.py
+++ b/app/services/user_db_service.py
@@ -3,17 +3,27 @@
 import importlib
 import sqlite3
 from typing import Any
-
 from fastapi import Depends
 
 from app.core.enum.db_driver import DBTypesEnum
 from app.core.exceptions import APIException
 from app.core.status import CommonCode
 from app.repository.user_db_repository import UserDbRepository, user_db_repository
-from app.schemas.user_db.result_model import BasicResult, SaveProfileResult, AllDBProfileResult
-from app.schemas.user_db.db_profile_model import DBProfileInfo, SaveDBProfile
 from app.core.utils import generate_prefixed_uuid
 from app.core.enum.db_key_prefix_name import DBSaveIdEnum
+from app.schemas.user_db.result_model import (
+    BasicResult,
+    SaveProfileResult,
+    AllDBProfileResult,
+    TableListResult,
+    ColumnListResult,
+    SchemaInfoResult
+)
+from app.schemas.user_db.db_profile_model import (
+    DBProfileInfo,
+    SaveDBProfile,
+    AllDBProfileInfo
+)
 
 user_db_repository_dependency = Depends(lambda: user_db_repository)
 
@@ -60,6 +70,90 @@ def find_all_profile(
         except Exception as e:
             raise APIException(CommonCode.FAIL) from e
 
+    def find_profile(
+        self,
+        profile_id,
+        repository: UserDbRepository = user_db_repository
+    ) -> AllDBProfileInfo:
+        """
+        특정 DB 연결 정보를 반환합니다.
+        """
+        try:
+            return repository.find_profile(profile_id)
+        except Exception as e:
+            raise APIException(CommonCode.FAIL) from e
+
+    def find_schemas(
+        self,
+        db_info: AllDBProfileInfo,
+        repository: UserDbRepository = user_db_repository
+    ) -> SchemaInfoResult:
+        """
+        DB 스키마 정보를 조회를 수행합니다.
+        """
+        try:
+            driver_module = self._get_driver_module(db_info.type)
+            connect_kwargs = self._prepare_connection_args(db_info)
+            schema_query = self._get_schema_query(db_info.type)
+
+            return repository.find_schemas(
+                driver_module,
+                schema_query,
+                **connect_kwargs
+            )
+        except Exception as e:
+            raise APIException(CommonCode.FAIL) from e
+
+    def find_tables(
+        self,
+        db_info: AllDBProfileInfo,
+        schema_name: str,
+        repository: UserDbRepository = user_db_repository
+    ) -> TableListResult:
+        """
+        특정 스키마 내의 테이블 정보를 조회합니다.
+        """
+        try:
+            driver_module = self._get_driver_module(db_info.type)
+            connect_kwargs = self._prepare_connection_args(db_info)
+            table_query = self._get_table_query(db_info.type, for_all_schemas=False)
+
+            return repository.find_tables(
+                driver_module,
+                table_query,
+                schema_name,
+                **connect_kwargs
+            )
+        except Exception as e:
+            raise APIException(CommonCode.FAIL) from e
+
+    def find_columns(
+        self,
+        db_info: AllDBProfileInfo,
+        schema_name: str,
+        table_name: str,
+        repository: UserDbRepository = user_db_repository
+    ) -> ColumnListResult:
+        """
+        특정 컬럼 정보를 조회합니다.
+        """
+        try:
+            driver_module = self._get_driver_module(db_info.type)
+            connect_kwargs = self._prepare_connection_args(db_info)
+            column_query = self._get_column_query(db_info.type)
+            db_type = db_info.type
+
+            return repository.find_columns(
+                driver_module,
+                column_query,
+                schema_name,
+                db_type,
+                table_name,
+                **connect_kwargs
+            )
+        except Exception as e:
+            raise APIException(CommonCode.FAIL) from e
+
     def _get_driver_module(self, db_type: str):
         """
         DB 타입에 따라 동적으로 드라이버 모듈을 로드합니다.
@@ -106,5 +200,77 @@ def _prepare_connection_args(self, db_info: DBProfileInfo) -> dict[str, Any]:
 
         return kwargs
 
+    def _get_schema_query(self, db_type: str) -> str | None:
+        db_type = db_type.lower()
+        if db_type == "postgresql":
+            return """
+                SELECT schema_name FROM information_schema.schemata
+                WHERE schema_name NOT IN ('pg_catalog', 'information_schema')
+            """
+        elif db_type in ["mysql", "mariadb"]:
+            return "SELECT schema_name FROM information_schema.schemata"
+        elif db_type == "oracle":
+            return "SELECT username FROM all_users"
+        elif db_type == "sqlite":
+            return None
+        return None
+
+
+    def _get_table_query(self, db_type: str, for_all_schemas: bool = False) -> str | None:  # 수정됨
+        db_type = db_type.lower()
+        if db_type == "postgresql":
+            if for_all_schemas:
+                return """
+                    SELECT table_name, table_schema FROM information_schema.tables
+                    WHERE table_type = 'BASE TABLE' AND table_schema NOT IN ('pg_catalog', 'information_schema')
+                """
+            else:
+                return """
+                    SELECT table_name, table_schema FROM information_schema.tables
+                    WHERE table_type = 'BASE TABLE' AND table_schema = %s
+                """
+        elif db_type in ["mysql", "mariadb"]:
+            if for_all_schemas:
+                return """
+                    SELECT table_name, table_schema FROM information_schema.tables
+                    WHERE table_type = 'BASE TABLE'
+                """
+            else:
+                return """
+                    SELECT table_name, table_schema FROM information_schema.tables
+                    WHERE table_type = 'BASE TABLE' AND table_schema = %s
+                """
+        elif db_type == "oracle":
+            return "SELECT table_name FROM all_tables WHERE owner = :owner"
+        elif db_type == "sqlite":
+            return "SELECT name FROM sqlite_master WHERE type='table'"
+        return None
+
+
+    def _get_column_query(self, db_type: str) -> str | None:
+        db_type = db_type.lower()
+        if db_type == "postgresql":
+            return """
+                SELECT column_name, data_type, is_nullable, column_default, table_name, table_schema
+                FROM information_schema.columns
+                WHERE table_schema NOT IN ('pg_catalog', 'information_schema')
+                AND table_schema = %s
+                AND table_name = %s
+            """
+        elif db_type in ["mysql", "mariadb"]:
+            return """
+                SELECT column_name, data_type, is_nullable, column_default, table_name, table_schema
+                FROM information_schema.columns
+                WHERE table_schema = %s AND table_name = %s
+            """
+        elif db_type == "oracle":
+            return """
+                SELECT column_name, data_type, nullable, data_default, table_name
+                FROM all_tab_columns
+                WHERE owner = :owner AND table_name = :table
+            """
+        elif db_type == "sqlite":
+            return None
+        return None
 
 user_db_service = UserDbService()

From 21930f91a4afa2beeecefc758239a4528faa5cf9 Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sun, 10 Aug 2025 18:34:53 +0900
Subject: [PATCH 15/25] =?UTF-8?q?feat:=20=EC=97=85=EB=8D=B0=EC=9D=B4?=
 =?UTF-8?q?=ED=8A=B8=20=EC=8B=9C=20=EC=82=AC=EC=9A=A9=EB=90=98=EB=8A=94=20?=
 =?UTF-8?q?=EC=83=81=ED=83=9C=20=EA=B0=92=20=EC=B6=94=EA=B0=80?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/core/status.py | 12 +++++++-----
 1 file changed, 7 insertions(+), 5 deletions(-)

diff --git a/app/core/status.py b/app/core/status.py
index 2f3f383..5129a23 100644
--- a/app/core/status.py
+++ b/app/core/status.py
@@ -25,6 +25,7 @@ class CommonCode(Enum):
     SUCCESS_FIND_TABLES = (status.HTTP_200_OK, "2104", "디비 테이블 정보 조회를 성공하였습니다.")
     SUCCESS_FIND_COLUMNS = (status.HTTP_200_OK, "2105", "디비 컬럼 정보 조회를 성공하였습니다.")
     SUCCESS_SAVE_DB_PROFILE = (status.HTTP_200_OK, "2130", "디비 연결 정보를 저장하였습니다.")
+    SUCCESS_UPDATE_DB_PROFILE = (status.HTTP_200_OK, "2150", "디비 연결 정보를 업데이트 하였습니다.")
 
 
     """ KEY 성공 코드 - 22xx """
@@ -93,11 +94,12 @@ class CommonCode(Enum):
 
     """ DRIVER, DB 서버 에러 코드 - 51xx """
     FAIL_CONNECT_DB = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5100", "디비 연결 중 에러가 발생했습니다.")
-    FAIL_SAVE_PROFILE = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5101", "디비 정보 저장 중 에러가 발생했습니다.")
-    FAIL_FIND_PROFILE = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5102", "디비 정보 조회 중 에러가 발생했습니다.")
-    FAIL_FIND_SCHEMAS = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5103", "디비 스키마 정보 조회 중 에러가 발생했습니다.")
-    FAIL_FIND_TABLES = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5104", "디비 테이블 정보 조회 중 에러가 발생했습니다.")
-    FAIL_FIND_COLUMNS = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5105", "디비 컬럼 정보 조회 중 에러가 발생했습니다.")
+    FAIL_FIND_PROFILE = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5101", "디비 정보 조회 중 에러가 발생했습니다.")
+    FAIL_FIND_SCHEMAS = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5102", "디비 스키마 정보 조회 중 에러가 발생했습니다.")
+    FAIL_FIND_TABLES = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5103", "디비 테이블 정보 조회 중 에러가 발생했습니다.")
+    FAIL_FIND_COLUMNS = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5104", "디비 컬럼 정보 조회 중 에러가 발생했습니다.")
+    FAIL_SAVE_PROFILE = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5130", "디비 정보 저장 중 에러가 발생했습니다.")
+    FAIL_UPDATE_PROFILE = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5150", "디비 정보 업데이트 중 에러가 발생했습니다.")
 
     """ KEY 서버 에러 코드 - 52xx """
 

From 27ebc1f048543f1c850bc34a15301aa458433bc0 Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sun, 10 Aug 2025 18:35:35 +0900
Subject: [PATCH 16/25] =?UTF-8?q?refactor:=20save=20=EC=8B=9C=20=EC=82=AC?=
 =?UTF-8?q?=EC=9A=A9=EB=90=98=EB=8A=94=20=EB=AA=A8=EB=8D=B8=20=EA=B3=B5?=
 =?UTF-8?q?=ED=86=B5=EC=9C=BC=EB=A1=9C=20=EC=82=AC=EC=9A=A9=ED=95=98?=
 =?UTF-8?q?=EB=8F=84=EB=A1=9D=20=EC=88=98=EC=A0=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/schemas/user_db/db_profile_model.py | 2 +-
 app/schemas/user_db/result_model.py     | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/app/schemas/user_db/db_profile_model.py b/app/schemas/user_db/db_profile_model.py
index ad312b1..8a1d813 100644
--- a/app/schemas/user_db/db_profile_model.py
+++ b/app/schemas/user_db/db_profile_model.py
@@ -52,7 +52,7 @@ def _is_empty(value: Any | None) -> bool:
             return True
         return False
 
-class SaveDBProfile(DBProfileInfo):
+class UpdateOrSaveDBProfile(DBProfileInfo):
     id: str | None = Field(None, description="DB Key 값")
     view_name: str | None = Field(None, description="DB 노출명")
 
diff --git a/app/schemas/user_db/result_model.py b/app/schemas/user_db/result_model.py
index 7b8f366..bd08364 100644
--- a/app/schemas/user_db/result_model.py
+++ b/app/schemas/user_db/result_model.py
@@ -12,7 +12,7 @@ class BasicResult(BaseModel):
     code: CommonCode = Field(None, description="결과 코드")
 
 # 디비 정보 후 반환되는 저장 모델
-class SaveProfileResult(BasicResult):
+class UpdateOrSaveProfileResult(BasicResult):
     """DB 조회 결과를 위한 확장 모델"""
     view_name: str = Field(..., description="저장된 디비명")
 

From 08ad1e56de02881ff1bae09ba3701080424a864b Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sun, 10 Aug 2025 18:35:55 +0900
Subject: [PATCH 17/25] =?UTF-8?q?feat:=20profile=20=EC=A0=95=EB=B3=B4=20?=
 =?UTF-8?q?=EC=97=85=EB=8D=B0=EC=9D=B4=ED=8A=B8=20=ED=95=98=EB=8A=94=20?=
 =?UTF-8?q?=EB=A1=9C=EC=A7=81=20=EC=83=9D=EC=84=B1?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/api/user_db_api.py               | 21 ++++++++++--
 app/repository/user_db_repository.py | 51 +++++++++++++++++++++++-----
 app/services/user_db_service.py      | 21 +++++++++---
 3 files changed, 79 insertions(+), 14 deletions(-)

diff --git a/app/api/user_db_api.py b/app/api/user_db_api.py
index e1bf203..3ce69cb 100644
--- a/app/api/user_db_api.py
+++ b/app/api/user_db_api.py
@@ -5,7 +5,7 @@
 
 from app.core.exceptions import APIException
 from app.core.response import ResponseMessage
-from app.schemas.user_db.db_profile_model import DBProfileInfo, SaveDBProfile
+from app.schemas.user_db.db_profile_model import DBProfileInfo, UpdateOrSaveDBProfile
 from app.services.user_db_service import UserDbService, user_db_service
 from app.schemas.user_db.result_model import DBProfile, TableInfo, ColumnInfo, SchemaListResult, TableListResult, ColumnListResult
 
@@ -36,7 +36,7 @@ def connection_test(
     summary="DB 프로필 저장",
 )
 def save_profile(
-    save_db_info: SaveDBProfile,
+    save_db_info: UpdateOrSaveDBProfile,
     service: UserDbService = user_db_service_dependency,
 ) -> ResponseMessage[str]:
 
@@ -47,6 +47,23 @@ def save_profile(
         raise APIException(result.code)
     return ResponseMessage.success(value=result.view_name, code=result.code)
 
+@router.put(
+    "/modify/profile",
+    response_model=ResponseMessage[str],
+    summary="DB 프로필 저장",
+)
+def modify_profile(
+    modify_db_info: UpdateOrSaveDBProfile,
+    service: UserDbService = user_db_service_dependency,
+) -> ResponseMessage[str]:
+
+    modify_db_info.validate_required_fields()
+    result = service.modify_profile(modify_db_info)
+
+    if not result.is_successful:
+        raise APIException(result.code)
+    return ResponseMessage.success(value=result.view_name, code=result.code)
+
 @router.get(
     "/find/all",
     response_model=ResponseMessage[List[DBProfile]],
diff --git a/app/repository/user_db_repository.py b/app/repository/user_db_repository.py
index 2f156c0..16d2e3c 100644
--- a/app/repository/user_db_repository.py
+++ b/app/repository/user_db_repository.py
@@ -9,7 +9,7 @@
 from app.schemas.user_db.result_model import (
     DBProfile,
     BasicResult,
-    SaveProfileResult,
+    UpdateOrSaveProfileResult,
     AllDBProfileResult,
     SchemaListResult,
     TableListResult,
@@ -17,7 +17,7 @@
     ColumnInfo
 )
 from app.schemas.user_db.db_profile_model import (
-    SaveDBProfile,
+    UpdateOrSaveDBProfile,
     AllDBProfileInfo
 )
 
@@ -57,10 +57,10 @@ def connection_test(
 
     def save_profile(
         self,
-        save_db_info: SaveDBProfile
-    ) -> SaveProfileResult:
+        save_db_info: UpdateOrSaveDBProfile
+    ) -> UpdateOrSaveProfileResult:
         """
-        DB 드라이버와 연결에 필요한 매개변수들을 받아 연결을 테스트합니다.
+        DB 드라이버와 연결에 필요한 매개변수들을 받아 저장합니다.
         """
         db_path = get_db_path()
         connection = None
@@ -83,11 +83,46 @@ def save_profile(
             connection.commit()
             name = save_db_info.view_name if save_db_info.view_name else save_db_info.type
 
-            return SaveProfileResult(is_successful=True, code=CommonCode.SUCCESS_SAVE_DB_PROFILE, view_name=name)
+            return UpdateOrSaveProfileResult(is_successful=True, code=CommonCode.SUCCESS_SAVE_DB_PROFILE, view_name=name)
+        except sqlite3.Error:
+            return UpdateOrSaveProfileResult(is_successful=False, code=CommonCode.FAIL_SAVE_PROFILE)
+        except Exception:
+            return UpdateOrSaveProfileResult(is_successful=False, code=CommonCode.FAIL_SAVE_PROFILE)
+        finally:
+            if connection:
+                connection.close()
+
+    def modify_profile(
+            self,
+            modify_db_info: UpdateOrSaveDBProfile
+    ) -> UpdateOrSaveProfileResult:
+        """
+        DB 드라이버와 연결에 필요한 매개변수들을 받아 업데이트합니다.
+        """
+        db_path = get_db_path()
+        connection = None
+        try:
+            connection = sqlite3.connect(db_path)
+            cursor = connection.cursor()
+            profile_dict = modify_db_info.model_dump()
+
+            columns_to_update = {
+                key: value for key, value in profile_dict.items() if value is not None and key != 'id'
+            }
+
+            set_clause = ", ".join([f"{key} = ?" for key in columns_to_update.keys()])
+            sql = f"UPDATE db_profile SET {set_clause} WHERE id = ?"
+            data_to_update = tuple(columns_to_update.values()) + (modify_db_info.id,)
+
+            cursor.execute(sql, data_to_update)
+            connection.commit()
+            name = modify_db_info.view_name if modify_db_info.view_name else modify_db_info.type
+
+            return UpdateOrSaveProfileResult(is_successful=True, code=CommonCode.SUCCESS_UPDATE_DB_PROFILE, view_name=name)
         except sqlite3.Error:
-            return SaveProfileResult(is_successful=False, code=CommonCode.FAIL_SAVE_PROFILE)
+            return UpdateOrSaveProfileResult(is_successful=False, code=CommonCode.FAIL_UPDATE_PROFILE)
         except Exception:
-            return SaveProfileResult(is_successful=False, code=CommonCode.FAIL_SAVE_PROFILE)
+            return UpdateOrSaveProfileResult(is_successful=False, code=CommonCode.FAIL_UPDATE_PROFILE)
         finally:
             if connection:
                 connection.close()
diff --git a/app/services/user_db_service.py b/app/services/user_db_service.py
index ead22e7..8015a89 100644
--- a/app/services/user_db_service.py
+++ b/app/services/user_db_service.py
@@ -13,7 +13,7 @@
 from app.core.enum.db_key_prefix_name import DBSaveIdEnum
 from app.schemas.user_db.result_model import (
     BasicResult,
-    SaveProfileResult,
+    UpdateOrSaveProfileResult,
     AllDBProfileResult,
     TableListResult,
     ColumnListResult,
@@ -21,7 +21,7 @@
 )
 from app.schemas.user_db.db_profile_model import (
     DBProfileInfo,
-    SaveDBProfile,
+    UpdateOrSaveDBProfile,
     AllDBProfileInfo
 )
 
@@ -46,9 +46,9 @@ def connection_test(
 
     def save_profile(
         self,
-        save_db_info: SaveDBProfile,
+        save_db_info: UpdateOrSaveDBProfile,
         repository: UserDbRepository = user_db_repository
-    ) -> SaveProfileResult:
+    ) -> UpdateOrSaveProfileResult:
         """
         DB 연결 정보를 저장 후 결과를 반환합니다.
         """
@@ -58,6 +58,19 @@ def save_profile(
         except Exception as e:
             raise APIException(CommonCode.FAIL) from e
 
+    def modify_profile(
+        self,
+        modify_db_info: UpdateOrSaveDBProfile,
+        repository: UserDbRepository = user_db_repository
+    ) -> UpdateOrSaveProfileResult:
+        """
+        DB 연결 정보를 업데이트 후 결과를 반환합니다.
+        """
+        try:
+            return repository.modify_profile(modify_db_info)
+        except Exception as e:
+            raise APIException(CommonCode.FAIL) from e
+
     def find_all_profile(
         self,
         repository: UserDbRepository = user_db_repository

From 61f0e7135b952aeebf6daec9e944935ee38a32ee Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sun, 10 Aug 2025 18:51:32 +0900
Subject: [PATCH 18/25] =?UTF-8?q?feat:=20=EC=A0=9C=EA=B1=B0=20=EC=8B=9C=20?=
 =?UTF-8?q?=EC=82=AC=EC=9A=A9=EB=90=98=EB=8A=94=20=EC=83=81=ED=83=9C=20?=
 =?UTF-8?q?=EA=B0=92=20=EC=B6=94=EA=B0=80?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/core/status.py | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/app/core/status.py b/app/core/status.py
index 5129a23..1dc25c5 100644
--- a/app/core/status.py
+++ b/app/core/status.py
@@ -24,8 +24,9 @@ class CommonCode(Enum):
     SUCCESS_FIND_SCHEMAS = (status.HTTP_200_OK, "2103", "디비 스키마 정보 조회를 성공하였습니다.")
     SUCCESS_FIND_TABLES = (status.HTTP_200_OK, "2104", "디비 테이블 정보 조회를 성공하였습니다.")
     SUCCESS_FIND_COLUMNS = (status.HTTP_200_OK, "2105", "디비 컬럼 정보 조회를 성공하였습니다.")
-    SUCCESS_SAVE_DB_PROFILE = (status.HTTP_200_OK, "2130", "디비 연결 정보를 저장하였습니다.")
-    SUCCESS_UPDATE_DB_PROFILE = (status.HTTP_200_OK, "2150", "디비 연결 정보를 업데이트 하였습니다.")
+    SUCCESS_SAVE_PROFILE = (status.HTTP_200_OK, "2130", "디비 연결 정보를 저장하였습니다.")
+    SUCCESS_UPDATE_PROFILE = (status.HTTP_200_OK, "2150", "디비 연결 정보를 업데이트 하였습니다.")
+    SUCCESS_DELETE_PROFILE = (status.HTTP_200_OK, "2170", "디비 연결 정보를 삭제 하였습니다.")
 
 
     """ KEY 성공 코드 - 22xx """
@@ -100,6 +101,7 @@ class CommonCode(Enum):
     FAIL_FIND_COLUMNS = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5104", "디비 컬럼 정보 조회 중 에러가 발생했습니다.")
     FAIL_SAVE_PROFILE = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5130", "디비 정보 저장 중 에러가 발생했습니다.")
     FAIL_UPDATE_PROFILE = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5150", "디비 정보 업데이트 중 에러가 발생했습니다.")
+    FAIL_DELETE_PROFILE = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5170", "디비 정보 삭제 중 에러가 발생했습니다.")
 
     """ KEY 서버 에러 코드 - 52xx """
 

From fdb01606279bd94d1196e1bb58610e432f24a254 Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sun, 10 Aug 2025 18:51:57 +0900
Subject: [PATCH 19/25] =?UTF-8?q?refactor:=20save=20=EB=B0=8F=20update=20?=
 =?UTF-8?q?=EC=8B=9C=20=EC=82=AC=EC=9A=A9=EB=90=98=EB=8A=94=20=EB=AA=A8?=
 =?UTF-8?q?=EB=8D=B8=20=EA=B3=B5=ED=86=B5=EC=9C=BC=EB=A1=9C=20=EC=82=AC?=
 =?UTF-8?q?=EC=9A=A9=ED=95=98=EB=8F=84=EB=A1=9D=20=EC=88=98=EC=A0=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/schemas/user_db/result_model.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/app/schemas/user_db/result_model.py b/app/schemas/user_db/result_model.py
index bd08364..28c375e 100644
--- a/app/schemas/user_db/result_model.py
+++ b/app/schemas/user_db/result_model.py
@@ -12,7 +12,7 @@ class BasicResult(BaseModel):
     code: CommonCode = Field(None, description="결과 코드")
 
 # 디비 정보 후 반환되는 저장 모델
-class UpdateOrSaveProfileResult(BasicResult):
+class ChangeProfileResult(BasicResult):
     """DB 조회 결과를 위한 확장 모델"""
     view_name: str = Field(..., description="저장된 디비명")
 

From 7c2b1e3d4a9eb6806e001653789a21941ac3bbcf Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sun, 10 Aug 2025 18:52:15 +0900
Subject: [PATCH 20/25] =?UTF-8?q?feat:=20=EC=82=AD=EC=A0=9C=20=ED=95=98?=
 =?UTF-8?q?=EB=8A=94=20=EB=A1=9C=EC=A7=81=20=EC=83=9D=EC=84=B1?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/api/user_db_api.py               | 18 ++++++++++-
 app/repository/user_db_repository.py | 45 ++++++++++++++++++++++------
 app/services/user_db_service.py      | 19 ++++++++++--
 3 files changed, 69 insertions(+), 13 deletions(-)

diff --git a/app/api/user_db_api.py b/app/api/user_db_api.py
index 3ce69cb..9a68f23 100644
--- a/app/api/user_db_api.py
+++ b/app/api/user_db_api.py
@@ -50,7 +50,7 @@ def save_profile(
 @router.put(
     "/modify/profile",
     response_model=ResponseMessage[str],
-    summary="DB 프로필 저장",
+    summary="DB 프로필 업데이트",
 )
 def modify_profile(
     modify_db_info: UpdateOrSaveDBProfile,
@@ -64,6 +64,22 @@ def modify_profile(
         raise APIException(result.code)
     return ResponseMessage.success(value=result.view_name, code=result.code)
 
+@router.delete(
+    "/remove/{profile_id}",
+    response_model=ResponseMessage[str],
+    summary="DB 프로필 삭제",
+)
+def remove_profile(
+    profile_id: str,
+    service: UserDbService = user_db_service_dependency,
+) -> ResponseMessage[str]:
+
+    result = service.remove_profile(profile_id)
+
+    if not result.is_successful:
+        raise APIException(result.code)
+    return ResponseMessage.success(value=result.view_name, code=result.code)
+
 @router.get(
     "/find/all",
     response_model=ResponseMessage[List[DBProfile]],
diff --git a/app/repository/user_db_repository.py b/app/repository/user_db_repository.py
index 16d2e3c..f0c065f 100644
--- a/app/repository/user_db_repository.py
+++ b/app/repository/user_db_repository.py
@@ -9,7 +9,7 @@
 from app.schemas.user_db.result_model import (
     DBProfile,
     BasicResult,
-    UpdateOrSaveProfileResult,
+    ChangeProfileResult,
     AllDBProfileResult,
     SchemaListResult,
     TableListResult,
@@ -58,7 +58,7 @@ def connection_test(
     def save_profile(
         self,
         save_db_info: UpdateOrSaveDBProfile
-    ) -> UpdateOrSaveProfileResult:
+    ) -> ChangeProfileResult:
         """
         DB 드라이버와 연결에 필요한 매개변수들을 받아 저장합니다.
         """
@@ -83,11 +83,11 @@ def save_profile(
             connection.commit()
             name = save_db_info.view_name if save_db_info.view_name else save_db_info.type
 
-            return UpdateOrSaveProfileResult(is_successful=True, code=CommonCode.SUCCESS_SAVE_DB_PROFILE, view_name=name)
+            return ChangeProfileResult(is_successful=True, code=CommonCode.SUCCESS_SAVE_PROFILE, view_name=name)
         except sqlite3.Error:
-            return UpdateOrSaveProfileResult(is_successful=False, code=CommonCode.FAIL_SAVE_PROFILE)
+            return ChangeProfileResult(is_successful=False, code=CommonCode.FAIL_SAVE_PROFILE)
         except Exception:
-            return UpdateOrSaveProfileResult(is_successful=False, code=CommonCode.FAIL_SAVE_PROFILE)
+            return ChangeProfileResult(is_successful=False, code=CommonCode.FAIL_SAVE_PROFILE)
         finally:
             if connection:
                 connection.close()
@@ -95,7 +95,7 @@ def save_profile(
     def modify_profile(
             self,
             modify_db_info: UpdateOrSaveDBProfile
-    ) -> UpdateOrSaveProfileResult:
+    ) -> ChangeProfileResult:
         """
         DB 드라이버와 연결에 필요한 매개변수들을 받아 업데이트합니다.
         """
@@ -118,11 +118,38 @@ def modify_profile(
             connection.commit()
             name = modify_db_info.view_name if modify_db_info.view_name else modify_db_info.type
 
-            return UpdateOrSaveProfileResult(is_successful=True, code=CommonCode.SUCCESS_UPDATE_DB_PROFILE, view_name=name)
+            return ChangeProfileResult(is_successful=True, code=CommonCode.SUCCESS_UPDATE_PROFILE, view_name=name)
         except sqlite3.Error:
-            return UpdateOrSaveProfileResult(is_successful=False, code=CommonCode.FAIL_UPDATE_PROFILE)
+            return ChangeProfileResult(is_successful=False, code=CommonCode.FAIL_UPDATE_PROFILE)
         except Exception:
-            return UpdateOrSaveProfileResult(is_successful=False, code=CommonCode.FAIL_UPDATE_PROFILE)
+            return ChangeProfileResult(is_successful=False, code=CommonCode.FAIL_UPDATE_PROFILE)
+        finally:
+            if connection:
+                connection.close()
+
+    def remove_profile(
+        self,
+        profile_id: str,
+    ) -> ChangeProfileResult:
+        """
+        DB 드라이버와 연결에 필요한 매개변수들을 받아 삭제합니다.
+        """
+        db_path = get_db_path()
+        connection = None
+        try:
+            connection = sqlite3.connect(db_path)
+            cursor = connection.cursor()
+
+            sql = "DELETE FROM db_profile WHERE id = ?"
+            data_to_delete = (profile_id,)
+
+            cursor.execute(sql, data_to_delete)
+            connection.commit()
+            return ChangeProfileResult(is_successful=True, code=CommonCode.SUCCESS_DELETE_PROFILE, view_name=profile_id)
+        except sqlite3.Error:
+            return ChangeProfileResult(is_successful=False, code=CommonCode.FAIL_DELETE_PROFILE)
+        except Exception:
+            return ChangeProfileResult(is_successful=False, code=CommonCode.FAIL_DELETE_PROFILE)
         finally:
             if connection:
                 connection.close()
diff --git a/app/services/user_db_service.py b/app/services/user_db_service.py
index 8015a89..da4361a 100644
--- a/app/services/user_db_service.py
+++ b/app/services/user_db_service.py
@@ -13,7 +13,7 @@
 from app.core.enum.db_key_prefix_name import DBSaveIdEnum
 from app.schemas.user_db.result_model import (
     BasicResult,
-    UpdateOrSaveProfileResult,
+    ChangeProfileResult,
     AllDBProfileResult,
     TableListResult,
     ColumnListResult,
@@ -48,7 +48,7 @@ def save_profile(
         self,
         save_db_info: UpdateOrSaveDBProfile,
         repository: UserDbRepository = user_db_repository
-    ) -> UpdateOrSaveProfileResult:
+    ) -> ChangeProfileResult:
         """
         DB 연결 정보를 저장 후 결과를 반환합니다.
         """
@@ -62,7 +62,7 @@ def modify_profile(
         self,
         modify_db_info: UpdateOrSaveDBProfile,
         repository: UserDbRepository = user_db_repository
-    ) -> UpdateOrSaveProfileResult:
+    ) -> ChangeProfileResult:
         """
         DB 연결 정보를 업데이트 후 결과를 반환합니다.
         """
@@ -71,6 +71,19 @@ def modify_profile(
         except Exception as e:
             raise APIException(CommonCode.FAIL) from e
 
+    def remove_profile(
+        self,
+        profile_id: str,
+        repository: UserDbRepository = user_db_repository
+    ) -> ChangeProfileResult:
+        """
+        DB 연결 정보를 삭제 후 결과를 반환합니다.
+        """
+        try:
+            return repository.remove_profile(profile_id)
+        except Exception as e:
+            raise APIException(CommonCode.FAIL) from e
+
     def find_all_profile(
         self,
         repository: UserDbRepository = user_db_repository

From e1188d8dd0380f7bf6674b136c1b0ff2342674f7 Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sun, 10 Aug 2025 19:22:20 +0900
Subject: [PATCH 21/25] =?UTF-8?q?style:=20modify=20=EB=B6=80=EB=B6=84=20?=
 =?UTF-8?q?=EB=84=A4=EC=9D=B4=EB=B0=8D=20=EC=9D=BC=EC=B9=98=EC=8B=9C?=
 =?UTF-8?q?=ED=82=A4=EB=8A=94=20=EC=9E=91=EC=97=85=20=EC=A7=84=ED=96=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/api/user_db_api.py               | 11 ++++++-----
 app/repository/user_db_repository.py | 10 +++++-----
 app/services/user_db_service.py      |  6 +++---
 3 files changed, 14 insertions(+), 13 deletions(-)

diff --git a/app/api/user_db_api.py b/app/api/user_db_api.py
index 9a68f23..d62fd4e 100644
--- a/app/api/user_db_api.py
+++ b/app/api/user_db_api.py
@@ -7,7 +7,7 @@
 from app.core.response import ResponseMessage
 from app.schemas.user_db.db_profile_model import DBProfileInfo, UpdateOrSaveDBProfile
 from app.services.user_db_service import UserDbService, user_db_service
-from app.schemas.user_db.result_model import DBProfile, TableInfo, ColumnInfo, SchemaListResult, TableListResult, ColumnListResult
+from app.schemas.user_db.result_model import DBProfile, ColumnInfo
 
 user_db_service_dependency = Depends(lambda: user_db_service)
 
@@ -23,6 +23,7 @@ def connection_test(
     db_info: DBProfileInfo,
     service: UserDbService = user_db_service_dependency,
 ) -> ResponseMessage[bool]:
+
     db_info.validate_required_fields()
     result = service.connection_test(db_info)
 
@@ -52,13 +53,13 @@ def save_profile(
     response_model=ResponseMessage[str],
     summary="DB 프로필 업데이트",
 )
-def modify_profile(
-    modify_db_info: UpdateOrSaveDBProfile,
+def update_profile(
+    update_db_info: UpdateOrSaveDBProfile,
     service: UserDbService = user_db_service_dependency,
 ) -> ResponseMessage[str]:
 
-    modify_db_info.validate_required_fields()
-    result = service.modify_profile(modify_db_info)
+    update_db_info.validate_required_fields()
+    result = service.update_profile(update_db_info)
 
     if not result.is_successful:
         raise APIException(result.code)
diff --git a/app/repository/user_db_repository.py b/app/repository/user_db_repository.py
index f0c065f..be45b99 100644
--- a/app/repository/user_db_repository.py
+++ b/app/repository/user_db_repository.py
@@ -92,9 +92,9 @@ def save_profile(
             if connection:
                 connection.close()
 
-    def modify_profile(
+    def update_profile(
             self,
-            modify_db_info: UpdateOrSaveDBProfile
+            update_db_info: UpdateOrSaveDBProfile
     ) -> ChangeProfileResult:
         """
         DB 드라이버와 연결에 필요한 매개변수들을 받아 업데이트합니다.
@@ -104,7 +104,7 @@ def modify_profile(
         try:
             connection = sqlite3.connect(db_path)
             cursor = connection.cursor()
-            profile_dict = modify_db_info.model_dump()
+            profile_dict = update_db_info.model_dump()
 
             columns_to_update = {
                 key: value for key, value in profile_dict.items() if value is not None and key != 'id'
@@ -112,11 +112,11 @@ def modify_profile(
 
             set_clause = ", ".join([f"{key} = ?" for key in columns_to_update.keys()])
             sql = f"UPDATE db_profile SET {set_clause} WHERE id = ?"
-            data_to_update = tuple(columns_to_update.values()) + (modify_db_info.id,)
+            data_to_update = tuple(columns_to_update.values()) + (update_db_info.id,)
 
             cursor.execute(sql, data_to_update)
             connection.commit()
-            name = modify_db_info.view_name if modify_db_info.view_name else modify_db_info.type
+            name = update_db_info.view_name if update_db_info.view_name else update_db_info.type
 
             return ChangeProfileResult(is_successful=True, code=CommonCode.SUCCESS_UPDATE_PROFILE, view_name=name)
         except sqlite3.Error:
diff --git a/app/services/user_db_service.py b/app/services/user_db_service.py
index da4361a..1189052 100644
--- a/app/services/user_db_service.py
+++ b/app/services/user_db_service.py
@@ -58,16 +58,16 @@ def save_profile(
         except Exception as e:
             raise APIException(CommonCode.FAIL) from e
 
-    def modify_profile(
+    def update_profile(
         self,
-        modify_db_info: UpdateOrSaveDBProfile,
+        update_db_info: UpdateOrSaveDBProfile,
         repository: UserDbRepository = user_db_repository
     ) -> ChangeProfileResult:
         """
         DB 연결 정보를 업데이트 후 결과를 반환합니다.
         """
         try:
-            return repository.modify_profile(modify_db_info)
+            return repository.update_profile(update_db_info)
         except Exception as e:
             raise APIException(CommonCode.FAIL) from e
 

From e8329fdd3913200fa7da3abab14c4018fb46eea2 Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sun, 10 Aug 2025 19:23:57 +0900
Subject: [PATCH 22/25] =?UTF-8?q?style:=20delete=20=EB=B6=80=EB=B6=84=20?=
 =?UTF-8?q?=EB=84=A4=EC=9D=B4=EB=B0=8D=20=EC=9D=BC=EA=B4=80=EC=84=B1=20?=
 =?UTF-8?q?=EC=9C=A0=EC=A7=80=ED=95=98=EB=8F=84=EB=A1=9D=20=EC=88=98?=
 =?UTF-8?q?=EC=A0=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/api/user_db_api.py               | 4 ++--
 app/repository/user_db_repository.py | 2 +-
 app/services/user_db_service.py      | 4 ++--
 3 files changed, 5 insertions(+), 5 deletions(-)

diff --git a/app/api/user_db_api.py b/app/api/user_db_api.py
index d62fd4e..06669bb 100644
--- a/app/api/user_db_api.py
+++ b/app/api/user_db_api.py
@@ -70,12 +70,12 @@ def update_profile(
     response_model=ResponseMessage[str],
     summary="DB 프로필 삭제",
 )
-def remove_profile(
+def delete_profile(
     profile_id: str,
     service: UserDbService = user_db_service_dependency,
 ) -> ResponseMessage[str]:
 
-    result = service.remove_profile(profile_id)
+    result = service.delete_profile(profile_id)
 
     if not result.is_successful:
         raise APIException(result.code)
diff --git a/app/repository/user_db_repository.py b/app/repository/user_db_repository.py
index be45b99..71c6a7f 100644
--- a/app/repository/user_db_repository.py
+++ b/app/repository/user_db_repository.py
@@ -127,7 +127,7 @@ def update_profile(
             if connection:
                 connection.close()
 
-    def remove_profile(
+    def delete_profile(
         self,
         profile_id: str,
     ) -> ChangeProfileResult:
diff --git a/app/services/user_db_service.py b/app/services/user_db_service.py
index 1189052..bf6855d 100644
--- a/app/services/user_db_service.py
+++ b/app/services/user_db_service.py
@@ -71,7 +71,7 @@ def update_profile(
         except Exception as e:
             raise APIException(CommonCode.FAIL) from e
 
-    def remove_profile(
+    def delete_profile(
         self,
         profile_id: str,
         repository: UserDbRepository = user_db_repository
@@ -80,7 +80,7 @@ def remove_profile(
         DB 연결 정보를 삭제 후 결과를 반환합니다.
         """
         try:
-            return repository.remove_profile(profile_id)
+            return repository.delete_profile(profile_id)
         except Exception as e:
             raise APIException(CommonCode.FAIL) from e
 

From 7dc13013e6028da340294e0489afab91c1dc1d62 Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sun, 10 Aug 2025 20:03:24 +0900
Subject: [PATCH 23/25] =?UTF-8?q?style:=20create=20=EB=B6=80=EB=B6=84=20?=
 =?UTF-8?q?=EB=84=A4=EC=9D=B4=EB=B0=8D=20=EC=9D=BC=EA=B4=80=EC=84=B1=20?=
 =?UTF-8?q?=EC=9E=88=EA=B2=8C=20=EC=88=98=EC=A0=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/api/user_db_api.py                  | 14 +++++++-------
 app/repository/user_db_repository.py    | 12 ++++++------
 app/schemas/user_db/db_profile_model.py |  2 +-
 app/services/user_db_service.py         | 12 ++++++------
 4 files changed, 20 insertions(+), 20 deletions(-)

diff --git a/app/api/user_db_api.py b/app/api/user_db_api.py
index 06669bb..81f6dbd 100644
--- a/app/api/user_db_api.py
+++ b/app/api/user_db_api.py
@@ -5,7 +5,7 @@
 
 from app.core.exceptions import APIException
 from app.core.response import ResponseMessage
-from app.schemas.user_db.db_profile_model import DBProfileInfo, UpdateOrSaveDBProfile
+from app.schemas.user_db.db_profile_model import DBProfileInfo, UpdateOrCreateDBProfile
 from app.services.user_db_service import UserDbService, user_db_service
 from app.schemas.user_db.result_model import DBProfile, ColumnInfo
 
@@ -32,17 +32,17 @@ def connection_test(
     return ResponseMessage.success(value=result.is_successful, code=result.code)
 
 @router.post(
-    "/save/profile",
+    "/create/profile",
     response_model=ResponseMessage[str],
     summary="DB 프로필 저장",
 )
-def save_profile(
-    save_db_info: UpdateOrSaveDBProfile,
+def create_profile(
+    create_db_info: UpdateOrCreateDBProfile,
     service: UserDbService = user_db_service_dependency,
 ) -> ResponseMessage[str]:
 
-    save_db_info.validate_required_fields()
-    result = service.save_profile(save_db_info)
+    create_db_info.validate_required_fields()
+    result = service.create_profile(create_db_info)
 
     if not result.is_successful:
         raise APIException(result.code)
@@ -54,7 +54,7 @@ def save_profile(
     summary="DB 프로필 업데이트",
 )
 def update_profile(
-    update_db_info: UpdateOrSaveDBProfile,
+    update_db_info: UpdateOrCreateDBProfile,
     service: UserDbService = user_db_service_dependency,
 ) -> ResponseMessage[str]:
 
diff --git a/app/repository/user_db_repository.py b/app/repository/user_db_repository.py
index 71c6a7f..340c760 100644
--- a/app/repository/user_db_repository.py
+++ b/app/repository/user_db_repository.py
@@ -17,7 +17,7 @@
     ColumnInfo
 )
 from app.schemas.user_db.db_profile_model import (
-    UpdateOrSaveDBProfile,
+    UpdateOrCreateDBProfile,
     AllDBProfileInfo
 )
 
@@ -55,9 +55,9 @@ def connection_test(
             if connection:
                 connection.close()
 
-    def save_profile(
+    def create_profile(
         self,
-        save_db_info: UpdateOrSaveDBProfile
+        create_db_info: UpdateOrCreateDBProfile
     ) -> ChangeProfileResult:
         """
         DB 드라이버와 연결에 필요한 매개변수들을 받아 저장합니다.
@@ -67,7 +67,7 @@ def save_profile(
         try:
             connection = sqlite3.connect(db_path)
             cursor = connection.cursor()
-            profile_dict = save_db_info.model_dump()
+            profile_dict = create_db_info.model_dump()
 
             columns_to_insert = {
                 key: value for key, value in profile_dict.items() if value is not None
@@ -81,7 +81,7 @@ def save_profile(
 
             cursor.execute(sql, data_to_insert)
             connection.commit()
-            name = save_db_info.view_name if save_db_info.view_name else save_db_info.type
+            name = create_db_info.view_name if create_db_info.view_name else create_db_info.type
 
             return ChangeProfileResult(is_successful=True, code=CommonCode.SUCCESS_SAVE_PROFILE, view_name=name)
         except sqlite3.Error:
@@ -94,7 +94,7 @@ def save_profile(
 
     def update_profile(
             self,
-            update_db_info: UpdateOrSaveDBProfile
+            update_db_info: UpdateOrCreateDBProfile
     ) -> ChangeProfileResult:
         """
         DB 드라이버와 연결에 필요한 매개변수들을 받아 업데이트합니다.
diff --git a/app/schemas/user_db/db_profile_model.py b/app/schemas/user_db/db_profile_model.py
index 8a1d813..f6d601f 100644
--- a/app/schemas/user_db/db_profile_model.py
+++ b/app/schemas/user_db/db_profile_model.py
@@ -52,7 +52,7 @@ def _is_empty(value: Any | None) -> bool:
             return True
         return False
 
-class UpdateOrSaveDBProfile(DBProfileInfo):
+class UpdateOrCreateDBProfile(DBProfileInfo):
     id: str | None = Field(None, description="DB Key 값")
     view_name: str | None = Field(None, description="DB 노출명")
 
diff --git a/app/services/user_db_service.py b/app/services/user_db_service.py
index bf6855d..5077b1f 100644
--- a/app/services/user_db_service.py
+++ b/app/services/user_db_service.py
@@ -21,7 +21,7 @@
 )
 from app.schemas.user_db.db_profile_model import (
     DBProfileInfo,
-    UpdateOrSaveDBProfile,
+    UpdateOrCreateDBProfile,
     AllDBProfileInfo
 )
 
@@ -44,23 +44,23 @@ def connection_test(
         except Exception as e:
             raise APIException(CommonCode.FAIL) from e
 
-    def save_profile(
+    def create_profile(
         self,
-        save_db_info: UpdateOrSaveDBProfile,
+        create_db_info: UpdateOrCreateDBProfile,
         repository: UserDbRepository = user_db_repository
     ) -> ChangeProfileResult:
         """
         DB 연결 정보를 저장 후 결과를 반환합니다.
         """
-        save_db_info.id = generate_prefixed_uuid(DBSaveIdEnum.user_db.value)
+        create_db_info.id = generate_prefixed_uuid(DBSaveIdEnum.user_db.value)
         try:
-            return repository.save_profile(save_db_info)
+            return repository.create_profile(create_db_info)
         except Exception as e:
             raise APIException(CommonCode.FAIL) from e
 
     def update_profile(
         self,
-        update_db_info: UpdateOrSaveDBProfile,
+        update_db_info: UpdateOrCreateDBProfile,
         repository: UserDbRepository = user_db_repository
     ) -> ChangeProfileResult:
         """

From da28d281516e1b9976214d1f615de485d32b78ec Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sun, 10 Aug 2025 20:09:19 +0900
Subject: [PATCH 24/25] =?UTF-8?q?refactor:=20=EC=97=B0=EA=B2=B0=20?=
 =?UTF-8?q?=ED=85=8C=EC=8A=A4=ED=8A=B8=EC=97=90=EC=84=9C=20=EB=A9=94?=
 =?UTF-8?q?=EC=84=9C=EB=93=9C=EB=A1=9C=20=EB=BA=80=20=EB=B6=80=EB=B6=84?=
 =?UTF-8?q?=EC=9D=84=20=EC=82=AC=EC=9A=A9=ED=95=98=EB=8F=84=EB=A1=9D=20?=
 =?UTF-8?q?=EC=88=98=EC=A0=95?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/repository/user_db_repository.py | 15 +--------------
 1 file changed, 1 insertion(+), 14 deletions(-)

diff --git a/app/repository/user_db_repository.py b/app/repository/user_db_repository.py
index 340c760..a6377ed 100644
--- a/app/repository/user_db_repository.py
+++ b/app/repository/user_db_repository.py
@@ -32,20 +32,7 @@ def connection_test(
         """
         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)
-
+            connection = self._connect(driver_module, **kwargs)
             return BasicResult(is_successful=True, code=CommonCode.SUCCESS_USER_DB_CONNECT_TEST)
         except (AttributeError, driver_module.OperationalError, driver_module.DatabaseError) as e:
             return BasicResult(is_successful=False, code=CommonCode.FAIL_CONNECT_DB)

From 6ad7745a75ba61571603c694a25da467b7eff6c5 Mon Sep 17 00:00:00 2001
From: ChoiseU <csw980928@gmail.com>
Date: Sun, 10 Aug 2025 20:45:37 +0900
Subject: [PATCH 25/25] =?UTF-8?q?refactor:=20=EC=BF=BC=EB=A6=AC=20?=
 =?UTF-8?q?=EC=83=9D=EC=84=B1=20=EB=B6=80=EB=B6=84=20service=20=EC=AA=BD?=
 =?UTF-8?q?=EC=9C=BC=EB=A1=9C=20=EB=B9=BC=EC=84=9C=20=EC=9E=91=EC=97=85?=
 =?UTF-8?q?=ED=95=98=EB=8F=84=EB=A1=9D=20=EA=B0=9C=EC=84=A0?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/repository/user_db_repository.py | 170 ++++++---------------------
 app/services/user_db_service.py      | 135 ++++++++++-----------
 2 files changed, 106 insertions(+), 199 deletions(-)

diff --git a/app/repository/user_db_repository.py b/app/repository/user_db_repository.py
index a6377ed..cd7a0e0 100644
--- a/app/repository/user_db_repository.py
+++ b/app/repository/user_db_repository.py
@@ -1,32 +1,26 @@
+import sqlite3
 from typing import Any
 
 import oracledb
-import sqlite3
 
+from app.core.exceptions import APIException
 from app.core.status import CommonCode
 from app.core.utils import get_db_path
-from app.core.exceptions import APIException
+from app.schemas.user_db.db_profile_model import AllDBProfileInfo, UpdateOrCreateDBProfile
 from app.schemas.user_db.result_model import (
-    DBProfile,
+    AllDBProfileResult,
     BasicResult,
     ChangeProfileResult,
-    AllDBProfileResult,
+    ColumnInfo,
+    ColumnListResult,
+    DBProfile,
     SchemaListResult,
     TableListResult,
-    ColumnListResult,
-    ColumnInfo
-)
-from app.schemas.user_db.db_profile_model import (
-    UpdateOrCreateDBProfile,
-    AllDBProfileInfo
 )
 
+
 class UserDbRepository:
-    def connection_test(
-        self,
-        driver_module: Any,
-        **kwargs: Any
-    ) -> BasicResult:
+    def connection_test(self, driver_module: Any, **kwargs: Any) -> BasicResult:
         """
         DB 드라이버와 연결에 필요한 매개변수들을 받아 연결을 테스트합니다.
         """
@@ -34,18 +28,15 @@ def connection_test(
         try:
             connection = self._connect(driver_module, **kwargs)
             return BasicResult(is_successful=True, code=CommonCode.SUCCESS_USER_DB_CONNECT_TEST)
-        except (AttributeError, driver_module.OperationalError, driver_module.DatabaseError) as e:
+        except (AttributeError, driver_module.OperationalError, driver_module.DatabaseError):
             return BasicResult(is_successful=False, code=CommonCode.FAIL_CONNECT_DB)
-        except Exception as e:
+        except Exception:
             return BasicResult(is_successful=False, code=CommonCode.FAIL)
         finally:
             if connection:
                 connection.close()
 
-    def create_profile(
-        self,
-        create_db_info: UpdateOrCreateDBProfile
-    ) -> ChangeProfileResult:
+    def create_profile(self, sql: str, data: tuple, create_db_info: UpdateOrCreateDBProfile) -> ChangeProfileResult:
         """
         DB 드라이버와 연결에 필요한 매개변수들을 받아 저장합니다.
         """
@@ -54,22 +45,9 @@ def create_profile(
         try:
             connection = sqlite3.connect(db_path)
             cursor = connection.cursor()
-            profile_dict = create_db_info.model_dump()
-
-            columns_to_insert = {
-                key: value for key, value in profile_dict.items() if value is not None
-            }
-
-            columns = ", ".join(columns_to_insert.keys())
-            placeholders = ", ".join(["?"] * len(columns_to_insert))
-
-            sql = f"INSERT INTO db_profile ({columns}) VALUES ({placeholders})"
-            data_to_insert = tuple(columns_to_insert.values())
-
-            cursor.execute(sql, data_to_insert)
+            cursor.execute(sql, data)
             connection.commit()
             name = create_db_info.view_name if create_db_info.view_name else create_db_info.type
-
             return ChangeProfileResult(is_successful=True, code=CommonCode.SUCCESS_SAVE_PROFILE, view_name=name)
         except sqlite3.Error:
             return ChangeProfileResult(is_successful=False, code=CommonCode.FAIL_SAVE_PROFILE)
@@ -79,10 +57,7 @@ def create_profile(
             if connection:
                 connection.close()
 
-    def update_profile(
-            self,
-            update_db_info: UpdateOrCreateDBProfile
-    ) -> ChangeProfileResult:
+    def update_profile(self, sql: str, data: tuple, update_db_info: UpdateOrCreateDBProfile) -> ChangeProfileResult:
         """
         DB 드라이버와 연결에 필요한 매개변수들을 받아 업데이트합니다.
         """
@@ -91,20 +66,9 @@ def update_profile(
         try:
             connection = sqlite3.connect(db_path)
             cursor = connection.cursor()
-            profile_dict = update_db_info.model_dump()
-
-            columns_to_update = {
-                key: value for key, value in profile_dict.items() if value is not None and key != 'id'
-            }
-
-            set_clause = ", ".join([f"{key} = ?" for key in columns_to_update.keys()])
-            sql = f"UPDATE db_profile SET {set_clause} WHERE id = ?"
-            data_to_update = tuple(columns_to_update.values()) + (update_db_info.id,)
-
-            cursor.execute(sql, data_to_update)
+            cursor.execute(sql, data)
             connection.commit()
             name = update_db_info.view_name if update_db_info.view_name else update_db_info.type
-
             return ChangeProfileResult(is_successful=True, code=CommonCode.SUCCESS_UPDATE_PROFILE, view_name=name)
         except sqlite3.Error:
             return ChangeProfileResult(is_successful=False, code=CommonCode.FAIL_UPDATE_PROFILE)
@@ -116,6 +80,8 @@ def update_profile(
 
     def delete_profile(
         self,
+        sql: str,
+        data: tuple,
         profile_id: str,
     ) -> ChangeProfileResult:
         """
@@ -126,11 +92,7 @@ def delete_profile(
         try:
             connection = sqlite3.connect(db_path)
             cursor = connection.cursor()
-
-            sql = "DELETE FROM db_profile WHERE id = ?"
-            data_to_delete = (profile_id,)
-
-            cursor.execute(sql, data_to_delete)
+            cursor.execute(sql, data)
             connection.commit()
             return ChangeProfileResult(is_successful=True, code=CommonCode.SUCCESS_DELETE_PROFILE, view_name=profile_id)
         except sqlite3.Error:
@@ -141,11 +103,9 @@ def delete_profile(
             if connection:
                 connection.close()
 
-    def find_all_profile(
-        self
-    ) -> AllDBProfileResult:
+    def find_all_profile(self, sql: str) -> AllDBProfileResult:
         """
-        모든 DB 연결 정보를 조회합니다.
+        전달받은 쿼리를 실행하여 모든 DB 연결 정보를 조회합니다.
         """
         db_path = get_db_path()
         connection = None
@@ -153,24 +113,9 @@ def find_all_profile(
             connection = sqlite3.connect(db_path)
             connection.row_factory = sqlite3.Row
             cursor = connection.cursor()
-
-            sql = """
-            SELECT
-                id,
-                type,
-                host,
-                port,
-                name,
-                username,
-                view_name,
-                created_at,
-                updated_at 
-            FROM db_profile
-            """
             cursor.execute(sql)
             rows = cursor.fetchall()
             profiles = [DBProfile(**row) for row in rows]
-
             return AllDBProfileResult(is_successful=True, code=CommonCode.SUCCESS_FIND_PROFILE, profiles=profiles)
         except sqlite3.Error:
             return AllDBProfileResult(is_successful=False, code=CommonCode.FAIL_FIND_PROFILE)
@@ -180,12 +125,9 @@ def find_all_profile(
             if connection:
                 connection.close()
 
-    def find_profile(
-        self,
-        profile_id
-    ) -> AllDBProfileInfo:
+    def find_profile(self, sql: str, data: tuple) -> AllDBProfileInfo:
         """
-        특정 DB 연결 정보를 조회합니다.
+        전달받은 쿼리를 실행하여 특정 DB 연결 정보를 조회합니다.
         """
         db_path = get_db_path()
         connection = None
@@ -193,30 +135,16 @@ def find_profile(
             connection = sqlite3.connect(db_path)
             connection.row_factory = sqlite3.Row
             cursor = connection.cursor()
-
-            sql = """
-            SELECT
-                id,
-                type,
-                host,
-                port,
-                name,
-                username,
-                password,
-                view_name,
-                created_at,
-                updated_at 
-            FROM db_profile
-            WHERE id = ?
-            """
-            cursor.execute(sql, (profile_id,))
+            cursor.execute(sql, data)
             row = cursor.fetchone()
 
+            if not row:
+                raise APIException(CommonCode.NO_SEARCH_DATA)
             return AllDBProfileInfo(**dict(row))
-        except sqlite3.Error:
-            raise APIException(CommonCode.FAIL_FIND_PROFILE)
-        except Exception:
-            raise APIException(CommonCode.FAIL)
+        except sqlite3.Error as e:
+            raise APIException(CommonCode.FAIL_FIND_PROFILE) from e
+        except Exception as e:
+            raise APIException(CommonCode.FAIL) from e
         finally:
             if connection:
                 connection.close()
@@ -224,23 +152,14 @@ def find_profile(
     # ─────────────────────────────
     # 스키마 조회
     # ─────────────────────────────
-    def find_schemas(
-            self,
-            driver_module: Any,
-            schema_query: str,
-            **kwargs: Any
-    ) -> SchemaListResult:
+    def find_schemas(self, driver_module: Any, schema_query: str, **kwargs: Any) -> SchemaListResult:
         connection = None
         try:
             connection = self._connect(driver_module, **kwargs)
             cursor = connection.cursor()
 
             if not schema_query:
-                return SchemaListResult(
-                    is_successful=True,
-                    code=CommonCode.SUCCESS_FIND_SCHEMAS,
-                    schemas=["main"]
-                )
+                return SchemaListResult(is_successful=True, code=CommonCode.SUCCESS_FIND_SCHEMAS, schemas=["main"])
 
             cursor.execute(schema_query)
             schemas = [row[0] for row in cursor.fetchall()]
@@ -255,13 +174,7 @@ def find_schemas(
     # ─────────────────────────────
     # 테이블 조회
     # ─────────────────────────────
-    def find_tables(
-        self,
-        driver_module: Any,
-        table_query: str,
-        schema_name: str,
-        **kwargs: Any
-    ) -> TableListResult:
+    def find_tables(self, driver_module: Any, table_query: str, schema_name: str, **kwargs: Any) -> TableListResult:
         connection = None
         try:
             connection = self._connect(driver_module, **kwargs)
@@ -287,13 +200,7 @@ def find_tables(
     # 컬럼 조회
     # ─────────────────────────────
     def find_columns(
-        self,
-        driver_module: Any,
-        column_query: str,
-        schema_name: str,
-        db_type,
-        table_name: str,
-        **kwargs: Any
+        self, driver_module: Any, column_query: str, schema_name: str, db_type, table_name: str, **kwargs: Any
     ) -> ColumnListResult:
         connection = None
         try:
@@ -312,7 +219,7 @@ def find_columns(
                         nullable=(c[3] == 0),  # notnull == 0 means nullable
                         default=c[4],
                         comment=None,
-                        is_pk=(c[5] == 1)
+                        is_pk=(c[5] == 1),
                     )
                     for c in columns_raw
                 ]
@@ -325,13 +232,11 @@ def find_columns(
                     try:
                         cursor.execute(column_query, {"owner": owner_bind, "table": table_bind})
                     except Exception:
-                        # fallback: try positional binds (:1, :2) if named binds fail
                         try:
                             pos_query = column_query.replace(":owner", ":1").replace(":table", ":2")
                             cursor.execute(pos_query, [owner_bind, table_bind])
-                        except Exception:
-                            # re-raise to be handled by outer exception handler
-                            raise
+                        except Exception as e:
+                            raise APIException(CommonCode.FAIL) from e
                 else:
                     cursor.execute(column_query)
 
@@ -342,7 +247,7 @@ def find_columns(
                         nullable=(c[2] in ["YES", "Y", True]),
                         default=c[3],
                         comment=c[4] if len(c) > 4 else None,
-                        is_pk=(c[5] in ["PRI", True] if len(c) > 5 else False)
+                        is_pk=(c[5] in ["PRI", True] if len(c) > 5 else False),
                     )
                     for c in cursor.fetchall()
                 ]
@@ -369,4 +274,5 @@ def _connect(self, driver_module: Any, **kwargs):
         else:
             return driver_module.connect(**kwargs)
 
+
 user_db_repository = UserDbRepository()
diff --git a/app/services/user_db_service.py b/app/services/user_db_service.py
index 5077b1f..9463d85 100644
--- a/app/services/user_db_service.py
+++ b/app/services/user_db_service.py
@@ -3,37 +3,30 @@
 import importlib
 import sqlite3
 from typing import Any
+
 from fastapi import Depends
 
 from app.core.enum.db_driver import DBTypesEnum
+from app.core.enum.db_key_prefix_name import DBSaveIdEnum
 from app.core.exceptions import APIException
 from app.core.status import CommonCode
-from app.repository.user_db_repository import UserDbRepository, user_db_repository
 from app.core.utils import generate_prefixed_uuid
-from app.core.enum.db_key_prefix_name import DBSaveIdEnum
+from app.repository.user_db_repository import UserDbRepository, user_db_repository
+from app.schemas.user_db.db_profile_model import AllDBProfileInfo, DBProfileInfo, UpdateOrCreateDBProfile
 from app.schemas.user_db.result_model import (
+    AllDBProfileResult,
     BasicResult,
     ChangeProfileResult,
-    AllDBProfileResult,
-    TableListResult,
     ColumnListResult,
-    SchemaInfoResult
-)
-from app.schemas.user_db.db_profile_model import (
-    DBProfileInfo,
-    UpdateOrCreateDBProfile,
-    AllDBProfileInfo
+    SchemaInfoResult,
+    TableListResult,
 )
 
 user_db_repository_dependency = Depends(lambda: user_db_repository)
 
 
 class UserDbService:
-    def connection_test(
-        self,
-        db_info: DBProfileInfo,
-        repository: UserDbRepository = user_db_repository
-    ) -> BasicResult:
+    def connection_test(self, db_info: DBProfileInfo, repository: UserDbRepository = user_db_repository) -> BasicResult:
         """
         DB 연결 정보를 받아 연결 테스트를 수행 후 결과를 반환합니다.
         """
@@ -45,74 +38,67 @@ def connection_test(
             raise APIException(CommonCode.FAIL) from e
 
     def create_profile(
-        self,
-        create_db_info: UpdateOrCreateDBProfile,
-        repository: UserDbRepository = user_db_repository
+        self, create_db_info: UpdateOrCreateDBProfile, repository: UserDbRepository = user_db_repository
     ) -> ChangeProfileResult:
         """
         DB 연결 정보를 저장 후 결과를 반환합니다.
         """
         create_db_info.id = generate_prefixed_uuid(DBSaveIdEnum.user_db.value)
         try:
-            return repository.create_profile(create_db_info)
+            # [수정] 쿼리와 데이터를 서비스에서 생성하여 레포지토리로 전달합니다.
+            sql, data = self._get_create_query_and_data(create_db_info)
+            return repository.create_profile(sql, data, create_db_info)
         except Exception as e:
             raise APIException(CommonCode.FAIL) from e
 
     def update_profile(
-        self,
-        update_db_info: UpdateOrCreateDBProfile,
-        repository: UserDbRepository = user_db_repository
+        self, update_db_info: UpdateOrCreateDBProfile, repository: UserDbRepository = user_db_repository
     ) -> ChangeProfileResult:
         """
         DB 연결 정보를 업데이트 후 결과를 반환합니다.
         """
         try:
-            return repository.update_profile(update_db_info)
+            # [수정] 쿼리와 데이터를 서비스에서 생성하여 레포지토리로 전달합니다.
+            sql, data = self._get_update_query_and_data(update_db_info)
+            return repository.update_profile(sql, data, update_db_info)
         except Exception as e:
             raise APIException(CommonCode.FAIL) from e
 
-    def delete_profile(
-        self,
-        profile_id: str,
-        repository: UserDbRepository = user_db_repository
-    ) -> ChangeProfileResult:
+    def delete_profile(self, profile_id: str, repository: UserDbRepository = user_db_repository) -> ChangeProfileResult:
         """
         DB 연결 정보를 삭제 후 결과를 반환합니다.
         """
         try:
-            return repository.delete_profile(profile_id)
+            # [수정] 쿼리와 데이터를 서비스에서 생성하여 레포지토리로 전달합니다.
+            sql, data = self._get_delete_query_and_data(profile_id)
+            return repository.delete_profile(sql, data, profile_id)
         except Exception as e:
             raise APIException(CommonCode.FAIL) from e
 
-    def find_all_profile(
-        self,
-        repository: UserDbRepository = user_db_repository
-    ) -> AllDBProfileResult:
+    def find_all_profile(self, repository: UserDbRepository = user_db_repository) -> AllDBProfileResult:
         """
         모든 DB 연결 정보를 반환합니다.
         """
         try:
-            return repository.find_all_profile()
+            # [수정] 쿼리를 서비스에서 생성하여 레포지토리로 전달합니다.
+            sql = self._get_find_all_query()
+            return repository.find_all_profile(sql)
         except Exception as e:
             raise APIException(CommonCode.FAIL) from e
 
-    def find_profile(
-        self,
-        profile_id,
-        repository: UserDbRepository = user_db_repository
-    ) -> AllDBProfileInfo:
+    def find_profile(self, profile_id, repository: UserDbRepository = user_db_repository) -> AllDBProfileInfo:
         """
         특정 DB 연결 정보를 반환합니다.
         """
         try:
-            return repository.find_profile(profile_id)
+            # [수정] 쿼리와 데이터를 서비스에서 생성하여 레포지토리로 전달합니다.
+            sql, data = self._get_find_one_query_and_data(profile_id)
+            return repository.find_profile(sql, data)
         except Exception as e:
             raise APIException(CommonCode.FAIL) from e
 
     def find_schemas(
-        self,
-        db_info: AllDBProfileInfo,
-        repository: UserDbRepository = user_db_repository
+        self, db_info: AllDBProfileInfo, repository: UserDbRepository = user_db_repository
     ) -> SchemaInfoResult:
         """
         DB 스키마 정보를 조회를 수행합니다.
@@ -122,19 +108,12 @@ def find_schemas(
             connect_kwargs = self._prepare_connection_args(db_info)
             schema_query = self._get_schema_query(db_info.type)
 
-            return repository.find_schemas(
-                driver_module,
-                schema_query,
-                **connect_kwargs
-            )
+            return repository.find_schemas(driver_module, schema_query, **connect_kwargs)
         except Exception as e:
             raise APIException(CommonCode.FAIL) from e
 
     def find_tables(
-        self,
-        db_info: AllDBProfileInfo,
-        schema_name: str,
-        repository: UserDbRepository = user_db_repository
+        self, db_info: AllDBProfileInfo, schema_name: str, repository: UserDbRepository = user_db_repository
     ) -> TableListResult:
         """
         특정 스키마 내의 테이블 정보를 조회합니다.
@@ -144,12 +123,7 @@ def find_tables(
             connect_kwargs = self._prepare_connection_args(db_info)
             table_query = self._get_table_query(db_info.type, for_all_schemas=False)
 
-            return repository.find_tables(
-                driver_module,
-                table_query,
-                schema_name,
-                **connect_kwargs
-            )
+            return repository.find_tables(driver_module, table_query, schema_name, **connect_kwargs)
         except Exception as e:
             raise APIException(CommonCode.FAIL) from e
 
@@ -158,7 +132,7 @@ def find_columns(
         db_info: AllDBProfileInfo,
         schema_name: str,
         table_name: str,
-        repository: UserDbRepository = user_db_repository
+        repository: UserDbRepository = user_db_repository,
     ) -> ColumnListResult:
         """
         특정 컬럼 정보를 조회합니다.
@@ -170,12 +144,7 @@ def find_columns(
             db_type = db_info.type
 
             return repository.find_columns(
-                driver_module,
-                column_query,
-                schema_name,
-                db_type,
-                table_name,
-                **connect_kwargs
+                driver_module, column_query, schema_name, db_type, table_name, **connect_kwargs
             )
         except Exception as e:
             raise APIException(CommonCode.FAIL) from e
@@ -241,7 +210,6 @@ def _get_schema_query(self, db_type: str) -> str | None:
             return None
         return None
 
-
     def _get_table_query(self, db_type: str, for_all_schemas: bool = False) -> str | None:  # 수정됨
         db_type = db_type.lower()
         if db_type == "postgresql":
@@ -272,7 +240,6 @@ def _get_table_query(self, db_type: str, for_all_schemas: bool = False) -> str |
             return "SELECT name FROM sqlite_master WHERE type='table'"
         return None
 
-
     def _get_column_query(self, db_type: str) -> str | None:
         db_type = db_type.lower()
         if db_type == "postgresql":
@@ -299,4 +266,38 @@ def _get_column_query(self, db_type: str) -> str | None:
             return None
         return None
 
+    # ─────────────────────────────
+    # 프로필 CRUD 쿼리 생성 메서드
+    # ─────────────────────────────
+    def _get_create_query_and_data(self, db_info: UpdateOrCreateDBProfile) -> tuple[str, tuple]:
+        profile_dict = db_info.model_dump()
+        columns_to_insert = {k: v for k, v in profile_dict.items() if v is not None}
+        columns = ", ".join(columns_to_insert.keys())
+        placeholders = ", ".join(["?"] * len(columns_to_insert))
+        sql = f"INSERT INTO db_profile ({columns}) VALUES ({placeholders})"
+        data = tuple(columns_to_insert.values())
+        return sql, data
+
+    def _get_update_query_and_data(self, db_info: UpdateOrCreateDBProfile) -> tuple[str, tuple]:
+        profile_dict = db_info.model_dump()
+        columns_to_update = {k: v for k, v in profile_dict.items() if v is not None and k != "id"}
+        set_clause = ", ".join([f"{key} = ?" for key in columns_to_update.keys()])
+        sql = f"UPDATE db_profile SET {set_clause} WHERE id = ?"
+        data = tuple(columns_to_update.values()) + (db_info.id,)
+        return sql, data
+
+    def _get_delete_query_and_data(self, profile_id: str) -> tuple[str, tuple]:
+        sql = "DELETE FROM db_profile WHERE id = ?"
+        data = (profile_id,)
+        return sql, data
+
+    def _get_find_all_query(self) -> str:
+        return "SELECT id, type, host, port, name, username, view_name, created_at, updated_at FROM db_profile"
+
+    def _get_find_one_query_and_data(self, profile_id: str) -> tuple[str, tuple]:
+        sql = "SELECT * FROM db_profile WHERE id = ?"
+        data = (profile_id,)
+        return sql, data
+
+
 user_db_service = UserDbService()