1차 개발 최종

app/api/annotation_api.py

@@ -2,6 +2,7 @@
 
 from app.core.response import ResponseMessage
 from app.core.status import CommonCode
+from app.schemas.annotation.hierarchical_response_model import HierarchicalDBMSAnnotation
 from app.schemas.annotation.request_model import AnnotationCreateRequest
 from app.schemas.annotation.response_model import AnnotationDeleteResponse, FullAnnotationResponse
 from app.services.annotation_service import AnnotationService, annotation_service
@@ -59,6 +60,22 @@ def get_annotation_by_db_profile_id(
     return ResponseMessage.success(value=annotation, code=CommonCode.SUCCESS_FIND_ANNOTATION)
 
 
+@router.get(
+    "/find/hierarchical/{db_profile_id}",
+    response_model=ResponseMessage[HierarchicalDBMSAnnotation],
+    summary="DB 프로필 ID로 계층적 어노테이션 조회",
+)
+def get_hierarchical_annotation_by_db_profile_id(
+    db_profile_id: str,
+    service: AnnotationService = annotation_service_dependency,
+) -> ResponseMessage[HierarchicalDBMSAnnotation]:
+    """
+    `db_profile_id`에 연결된 어노테이션을 계층 구조(DBMS > DB > 스키마 > 테이블 > 컬럼)로 조회합니다.
+    """
+    annotation = service.get_hierarchical_annotation_by_db_profile_id(db_profile_id)
+    return ResponseMessage.success(value=annotation, code=CommonCode.SUCCESS_FIND_ANNOTATION)
+
+
 @router.delete(
     "/remove/{annotation_id}",
     response_model=ResponseMessage[AnnotationDeleteResponse],

app/core/status.py

@@ -118,7 +118,12 @@ class CommonCode(Enum):
         "5002",
         "데이터 생성 후 검증 과정에서 에러가 발생했습니다.",
     )
-
+    FAIL_AI_SERVER_CONNECTION = (status.HTTP_503_SERVICE_UNAVAILABLE, "5003", "AI 서버 연결에 실패했습니다.")
+    FAIL_AI_SERVER_PROCESSING = (
+        status.HTTP_500_INTERNAL_SERVER_ERROR,
+        "5004",
+        "AI 서버가 요청을 처리하는 데 실패했습니다.",
+    )
     """ DRIVER, DB 서버 에러 코드 - 51xx """
     FAIL_CONNECT_DB = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5100", "디비 연결 중 에러가 발생했습니다.")
     FAIL_FIND_PROFILE = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5101", "디비 정보 조회 중 에러가 발생했습니다.")
@@ -154,12 +159,6 @@ class CommonCode(Enum):
     FAIL_CREATE_ANNOTATION = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5400", "어노테이션 생성 중 에러가 발생했습니다.")
     FAIL_FIND_ANNOTATION = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5401", "어노테이션 조회 중 에러가 발생했습니다.")
     FAIL_DELETE_ANNOTATION = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5402", "어노테이션 삭제 중 에러가 발생했습니다.")
-    FAIL_AI_SERVER_CONNECTION = (status.HTTP_503_SERVICE_UNAVAILABLE, "5403", "AI 서버 연결에 실패했습니다.")
-    FAIL_AI_SERVER_PROCESSING = (
-        status.HTTP_500_INTERNAL_SERVER_ERROR,
-        "5404",
-        "AI 서버가 요청을 처리하는 데 실패했습니다.",
-    )
 
     """ SQL 서버 에러 코드 - 55xx """
 

app/repository/annotation_repository.py

@@ -1,5 +1,8 @@
+import logging
 import sqlite3
 
+from app.core.exceptions import APIException
+from app.core.status import CommonCode
 from app.core.utils import get_db_path
 from app.schemas.annotation.db_model import (
     ColumnAnnotationInDB,
@@ -10,6 +13,13 @@
     TableAnnotationInDB,
     TableConstraintInDB,
 )
+from app.schemas.annotation.hierarchical_response_model import (
+    HierarchicalColumnAnnotation,
+    HierarchicalDBAnnotation,
+    HierarchicalDBMSAnnotation,
+    HierarchicalRelationshipAnnotation,
+    HierarchicalTableAnnotation,
+)
 from app.schemas.annotation.response_model import (
     ColumnAnnotationDetail,
     ConstraintDetail,
@@ -99,7 +109,7 @@ def create_full_annotation(
             (
                 c.id,
                 c.table_annotation_id,
-                c.constraint_type,
+                c.constraint_type.value,
                 c.name,
                 c.description,
                 c.expression,
@@ -279,6 +289,139 @@ def find_full_annotation_by_id(self, annotation_id: str) -> FullAnnotationRespon
             if conn:
                 conn.close()
 
+    def find_hierarchical_annotation_by_profile_id(self, db_profile_id: str) -> HierarchicalDBMSAnnotation | None:
+        """
+        db_profile_id로 계층적 어노테이션 정보를 조회합니다.
+        - DBMS > DB > 테이블 > 컬럼 구조로 데이터를 조립하여 반환합니다.
+        """
+        db_path = get_db_path()
+        conn = None
+        try:
+            conn = sqlite3.connect(str(db_path), timeout=10)
+            conn.row_factory = sqlite3.Row
+            cursor = conn.cursor()
+
+            # 1. 기본 정보 조회 (db_profile, database_annotation)
+            cursor.execute(
+                """
+                SELECT
+                    dp.type as dbms_type,
+                    da.id as annotation_id,
+                    da.db_profile_id,
+                    da.database_name,
+                    da.description as db_description,
+                    da.created_at,
+                    da.updated_at
+                FROM db_profile dp
+                JOIN database_annotation da ON dp.annotation_id = da.id
+                WHERE dp.id = ?
+                """,
+                (db_profile_id,),
+            )
+            base_info = cursor.fetchone()
+            if not base_info:
+                return None
+
+            # 2. 테이블 및 컬럼 정보 한번에 조회
+            cursor.execute(
+                """
+                SELECT
+                    ta.id as table_id,
+                    ta.table_name,
+                    ta.description as table_description,
+                    ca.column_name,
+                    ca.description as column_description,
+                    ca.data_type
+                FROM table_annotation ta
+                JOIN column_annotation ca ON ta.id = ca.table_annotation_id
+                WHERE ta.database_annotation_id = ?
+                ORDER BY ta.table_name, ca.ordinal_position
+                """,
+                (base_info["annotation_id"],),
+            )
+            rows = cursor.fetchall()
+
+            # 3. 데이터 계층 구조로 조립 (테이블, 컬럼)
+            tables_map = {}
+            for row in rows:
+                table_name = row["table_name"]
+                if table_name not in tables_map:
+                    tables_map[table_name] = HierarchicalTableAnnotation(
+                        table_name=table_name,
+                        description=row["table_description"],
+                        columns=[],
+                    )
+                tables_map[table_name].columns.append(
+                    HierarchicalColumnAnnotation(
+                        column_name=row["column_name"],
+                        description=row["column_description"],
+                        data_type=row["data_type"],
+                    )
+                )
+
+            # 4. 관계 정보 조회
+            cursor.execute(
+                """
+                SELECT
+                    ta_from.table_name as from_table,
+                    ca_from.column_name as from_column,
+                    tc.ref_table as to_table,
+                    cc.referenced_column_name as to_column,
+                    tc.name as constraint_name,
+                    tc.description as relationship_description
+                FROM table_constraint tc
+                JOIN table_annotation ta_from ON tc.table_annotation_id = ta_from.id
+                JOIN constraint_column cc ON tc.id = cc.constraint_id
+                JOIN column_annotation ca_from ON cc.column_annotation_id = ca_from.id
+                WHERE ta_from.database_annotation_id = ? AND tc.constraint_type = 'FOREIGN KEY'
+                ORDER BY tc.name, cc.position
+                """,
+                (base_info["annotation_id"],),
+            )
+            relationship_rows = cursor.fetchall()
+            logging.info(f"Raw relationship rows from DB: {[dict(row) for row in relationship_rows]}")
+
+            relationships_map = {}
+            for row in relationship_rows:
+                constraint_name = row["constraint_name"]
+                if constraint_name not in relationships_map:
+                    relationships_map[constraint_name] = {
+                        "from_table": row["from_table"],
+                        "to_table": row["to_table"],
+                        "description": row["relationship_description"],
+                        "from_columns": [],
+                        "to_columns": [],
+                    }
+                relationships_map[constraint_name]["from_columns"].append(row["from_column"])
+                relationships_map[constraint_name]["to_columns"].append(row["to_column"])
+
+            logging.info(f"Processed relationships map: {relationships_map}")
+            relationships = [HierarchicalRelationshipAnnotation(**data) for data in relationships_map.values()]
+            logging.info(f"Final relationships list: {relationships}")
+
+            # 5. 최종 데이터 조립
+            db = HierarchicalDBAnnotation(
+                db_name=base_info["database_name"],
+                description=base_info["db_description"],
+                tables=list(tables_map.values()),
+                relationships=relationships,
+            )
+
+            return HierarchicalDBMSAnnotation(
+                dbms_type=base_info["dbms_type"],
+                databases=[db],
+                annotation_id=base_info["annotation_id"],
+                db_profile_id=base_info["db_profile_id"],
+                created_at=base_info["created_at"],
+                updated_at=base_info["updated_at"],
+            )
+
+        except sqlite3.Error as e:
+            raise APIException(CommonCode.FAIL_FIND_ANNOTATION) from e
+        finally:
+            if conn:
+                conn.close()
+
     def delete_annotation_by_id(self, annotation_id: str) -> bool:
         """
         annotationId로 특정 어노테이션을 삭제합니다.

app/repository/chat_tab_repository.py

@@ -56,7 +56,7 @@ def updated_chat_tab(self, id: str, new_name: str | None) -> ChatTabInDB | None:
 
             # 데이터 업데이트
             cursor.execute(
-                "UPDATE chat_tab SET name = ?, updated_at = datetime('now', 'localtime') WHERE id = ?",
+                "UPDATE chat_tab SET name = ?, updated_at = datetime('now') WHERE id = ?",
                 (new_name, id),
             )
             conn.commit()
@@ -73,6 +73,20 @@ def updated_chat_tab(self, id: str, new_name: str | None) -> ChatTabInDB | None:
             if conn:
                 conn.close()
 
+    def update_tab_timestamp(self, id: str) -> bool:
+        """지정된 ID의 채팅 탭의 updated_at 타임스탬프를 현재 시간으로 업데이트합니다."""
+        db_path = get_db_path()
+        conn = None
+        try:
+            conn = sqlite3.connect(str(db_path), timeout=10)
+            cursor = conn.cursor()
+            cursor.execute("UPDATE chat_tab SET updated_at = datetime('now') WHERE id = ?", (id,))
+            conn.commit()
+            return cursor.rowcount > 0
+        finally:
+            if conn:
+                conn.close()
+
     def delete_chat_tab(self, id: str) -> bool:
         """채팅 탭ID에 해당하는 ChatTab을 삭제하고, 성공 여부를 반환합니다."""
         db_path = get_db_path()

app/schemas/annotation/hierarchical_response_model.py

@@ -0,0 +1,39 @@
+from datetime import datetime
+
+from pydantic import BaseModel, Field
+
+
+class HierarchicalColumnAnnotation(BaseModel):
+    column_name: str = Field(..., description="컬럼 이름")
+    description: str | None = Field(None, description="AI가 생성한 컬럼 어노테이션")
+    data_type: str | None = Field(None, description="컬럼 데이터 타입")
+
+
+class HierarchicalTableAnnotation(BaseModel):
+    table_name: str = Field(..., description="테이블 이름")
+    description: str | None = Field(None, description="AI가 생성한 테이블 어노테이션")
+    columns: list[HierarchicalColumnAnnotation] = Field(..., description="테이블에 속한 컬럼 목록")
+
+
+class HierarchicalRelationshipAnnotation(BaseModel):
+    from_table: str = Field(..., description="외래 키 제약조건이 시작되는 테이블")
+    from_columns: list[str] = Field(..., description="외래 키에 포함된 컬럼들")
+    to_table: str = Field(..., description="외래 키가 참조하는 테이블")
+    to_columns: list[str] = Field(..., description="참조되는 테이블의 컬럼들")
+    description: str | None = Field(None, description="AI가 생성한 관계 어노테이션")
+
+
+class HierarchicalDBAnnotation(BaseModel):
+    db_name: str = Field(..., description="데이터베이스 이름")
+    description: str | None = Field(None, description="AI가 생성한 데이터베이스 어노테이션")
+    tables: list[HierarchicalTableAnnotation] = Field(..., description="데이터베이스에 속한 테이블 목록")
+    relationships: list[HierarchicalRelationshipAnnotation] = Field(..., description="테이블 간의 관계 목록")
+
+
+class HierarchicalDBMSAnnotation(BaseModel):
+    dbms_type: str = Field(..., description="DBMS 종류 (e.g., postgresql, oracle)")
+    databases: list[HierarchicalDBAnnotation] = Field(..., description="DBMS에 속한 데이터베이스 목록")
+    annotation_id: str = Field(..., description="최상위 어노테이션 ID")
+    db_profile_id: str = Field(..., description="DB 프로필 ID")
+    created_at: datetime = Field(..., description="생성 일시")
+    updated_at: datetime = Field(..., description="수정 일시")

app/schemas/chat_message/base_model.py

@@ -29,4 +29,4 @@ def validate_chat_tab_id(self, field_value: str) -> None:
     def validate_message(self, field_value: str) -> None:
         """메시지 유효성 검사"""
         if not field_value or field_value.strip() == "":
-            raise APIException(CommonCode.INVALID_ANNOTATION_REQUEST)
+            raise APIException(CommonCode.INVALID_CHAT_MESSAGE_REQUEST)