[BAK-57] [refactor] DB & 어노테이션 ONE-TO-ONE

app/api/annotation_api.py

@@ -43,6 +43,22 @@ def get_annotation(
     return ResponseMessage.success(value=annotation, code=CommonCode.SUCCESS_FIND_ANNOTATION)
 
 
+@router.get(
+    "/find/db/{db_profile_id}",
+    response_model=ResponseMessage[FullAnnotationResponse],
+    summary="DB 프로필 ID로 어노테이션 조회",
+)
+def get_annotation_by_db_profile_id(
+    db_profile_id: str,
+    service: AnnotationService = annotation_service_dependency,
+) -> ResponseMessage[FullAnnotationResponse]:
+    """
+    `db_profile_id`에 연결된 어노테이션의 전체 상세 정보를 조회합니다.
+    """
+    annotation = service.get_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

@@ -59,6 +59,7 @@ class CommonCode(Enum):
     """ DRIVER, DB 클라이언트 에러 코드 - 41xx """
     INVALID_DB_DRIVER = (status.HTTP_409_CONFLICT, "4100", "지원하지 않는 데이터베이스입니다.")
     NO_DB_DRIVER = (status.HTTP_400_BAD_REQUEST, "4101", "데이터베이스는 필수 값입니다.")
+    NO_DB_PROFILE_FOUND = (status.HTTP_404_NOT_FOUND, "4102", "해당 ID의 DB 프로필을 찾을 수 없습니다.")
 
     """ KEY 클라이언트 에러 코드 - 42xx """
     INVALID_API_KEY_FORMAT = (status.HTTP_400_BAD_REQUEST, "4200", "API 키의 형식이 올바르지 않습니다.")
@@ -86,6 +87,7 @@ class CommonCode(Enum):
 
     """ ANNOTATION 클라이언트 에러 코드 - 44xx """
     INVALID_ANNOTATION_REQUEST = (status.HTTP_400_BAD_REQUEST, "4400", "어노테이션 요청 데이터가 유효하지 않습니다.")
+    NO_ANNOTATION_FOR_PROFILE = (status.HTTP_404_NOT_FOUND, "4401", "해당 DB 프로필에 연결된 어노테이션이 없습니다.")
 
     """ SQL 클라이언트 에러 코드 - 45xx """
 
@@ -116,6 +118,7 @@ class CommonCode(Enum):
         "5105",
         "디비 제약조건 또는 인덱스 정보 조회 중 에러가 발생했습니다.",
     )
+    FAIL_FIND_SAMPLE_ROWS = (status.HTTP_500_INTERNAL_SERVER_ERROR, "5106", "샘플 데이터 조회 중 에러가 발생했습니다.")
     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", "디비 정보 삭제 중 에러가 발생했습니다.")

app/db/init_db.py

@@ -80,14 +80,18 @@ def initialize_database():
             "username": "VARCHAR(128)",
             "password": "VARCHAR(128)",
             "view_name": "VARCHAR(64)",
+            "annotation_id": "VARCHAR(64)",
             "created_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP",
             "updated_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP",
+            "FOREIGN KEY (annotation_id)": "REFERENCES database_annotation(id) ON DELETE SET NULL",
         }
         create_sql = (
             f"CREATE TABLE IF NOT EXISTS db_profile ({', '.join([f'{k} {v}' for k, v in db_profile_cols.items()])})"
         )
         cursor.execute(create_sql)
-        _synchronize_table(cursor, "db_profile", db_profile_cols)
+        _synchronize_table(
+            cursor, "db_profile", {k: v for k, v in db_profile_cols.items() if not k.startswith("FOREIGN KEY")}
+        )
 
         cursor.execute(
             """
@@ -303,6 +307,7 @@ def initialize_database():
             "table_annotation_id": "VARCHAR(64) NOT NULL",
             "constraint_type": "VARCHAR(16) NOT NULL",
             "name": "VARCHAR(255)",
+            "description": "TEXT",
             "expression": "TEXT",
             "ref_table": "VARCHAR(255)",
             "on_update_action": "VARCHAR(16)",

app/repository/annotation_repository.py

@@ -20,6 +20,11 @@
 
 
 class AnnotationRepository:
+    """
+    어노테이션 데이터에 대한 데이터베이스 CRUD 작업을 처리합니다.
+    모든 메서드는 내부적으로 `sqlite3`를 사용하여 로컬 DB와 상호작용합니다.
+    """
+
     def create_full_annotation(
         self,
         db_conn: sqlite3.Connection,
@@ -96,6 +101,7 @@ def create_full_annotation(
                 c.table_annotation_id,
                 c.constraint_type,
                 c.name,
+                c.description,
                 c.expression,
                 c.ref_table,
                 c.on_update_action,
@@ -107,8 +113,8 @@ def create_full_annotation(
         ]
         cursor.executemany(
             """
-            INSERT INTO table_constraint (id, table_annotation_id, constraint_type, name, expression, ref_table, on_update_action, on_delete_action, created_at, updated_at)
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            INSERT INTO table_constraint (id, table_annotation_id, constraint_type, name, description, expression, ref_table, on_update_action, on_delete_action, created_at, updated_at)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
             """,
             constraint_data,
         )
@@ -155,6 +161,20 @@ def create_full_annotation(
             index_column_data,
         )
 
+    def update_db_profile_annotation_id(
+        self, db_conn: sqlite3.Connection, db_profile_id: str, annotation_id: str
+    ) -> None:
+        """
+        주어진 db_profile_id에 해당하는 레코드의 annotation_id를 업데이트합니다.
+        - 서비스 계층에서 트랜잭션을 관리하므로 connection을 인자로 받습니다.
+        - 실패 시 sqlite3.Error를 발생시킵니다.
+        """
+        cursor = db_conn.cursor()
+        cursor.execute(
+            "UPDATE db_profile SET annotation_id = ? WHERE id = ?",
+            (annotation_id, db_profile_id),
+        )
+
     def find_full_annotation_by_id(self, annotation_id: str) -> FullAnnotationResponse | None:
         """
         annotationId로 전체 어노테이션 상세 정보를 조회합니다.
@@ -182,29 +202,41 @@ def find_full_annotation_by_id(self, annotation_id: str) -> FullAnnotationRespon
 
                 # 컬럼 정보
                 cursor.execute(
-                    "SELECT id, column_name, description FROM column_annotation WHERE table_annotation_id = ?",
+                    "SELECT id, column_name, description, data_type, is_nullable, default_value FROM column_annotation WHERE table_annotation_id = ?",
                     (table_id,),
                 )
-                columns = [ColumnAnnotationDetail.model_validate(dict(c)) for c in cursor.fetchall()]
+                columns = []
+                for c in cursor.fetchall():
+                    c_dict = dict(c)
+                    c_dict["is_nullable"] = (
+                        bool(c_dict["is_nullable"]) if c_dict.get("is_nullable") is not None else None
+                    )
+                    columns.append(ColumnAnnotationDetail.model_validate(c_dict))
 
                 # 제약조건 정보
                 cursor.execute(
                     """
-                    SELECT tc.name, tc.constraint_type, ca.column_name
+                    SELECT tc.name, tc.constraint_type, tc.description, ca.column_name
                     FROM table_constraint tc
-                    JOIN constraint_column cc ON tc.id = cc.constraint_id
-                    JOIN column_annotation ca ON cc.column_annotation_id = ca.id
+                    LEFT JOIN constraint_column cc ON tc.id = cc.constraint_id
+                    LEFT JOIN column_annotation ca ON cc.column_annotation_id = ca.id
                     WHERE tc.table_annotation_id = ?
                     """,
                     (table_id,),
                 )
                 constraint_map = {}
                 for row in cursor.fetchall():
                     if row["name"] not in constraint_map:
-                        constraint_map[row["name"]] = {"type": row["constraint_type"], "columns": []}
-                    constraint_map[row["name"]]["columns"].append(row["column_name"])
+                        constraint_map[row["name"]] = {
+                            "type": row["constraint_type"],
+                            "columns": [],
+                            "description": row["description"],
+                        }
+                    if row["column_name"]:
+                        constraint_map[row["name"]]["columns"].append(row["column_name"])
                 constraints = [
-                    ConstraintDetail(name=k, type=v["type"], columns=v["columns"]) for k, v in constraint_map.items()
+                    ConstraintDetail(name=k, type=v["type"], columns=v["columns"], description=v["description"])
+                    for k, v in constraint_map.items()
                 ]
 
                 # 인덱스 정보