Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions app/db/init_db.py
Original file line number Diff line number Diff line change
Expand Up @@ -303,6 +303,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)",
Expand Down
38 changes: 28 additions & 10 deletions app/repository/annotation_repository.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,11 @@


class AnnotationRepository:
"""
어노테이션 데이터에 대한 데이터베이스 CRUD 작업을 처리합니다.
모든 메서드는 내부적으로 `sqlite3`를 사용하여 로컬 DB와 상호작용합니다.
"""

def create_full_annotation(
self,
db_conn: sqlite3.Connection,
Expand Down Expand Up @@ -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,
Expand All @@ -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,
)
Expand Down Expand Up @@ -182,29 +188,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()
]

# 인덱스 정보
Expand Down
Loading