diff --git a/app/api/api_key_api.py b/app/api/api_key_api.py index 74884ea..53ea9f2 100644 --- a/app/api/api_key_api.py +++ b/app/api/api_key_api.py @@ -63,7 +63,7 @@ def get_all_api_keys( ) for api_key in api_keys_in_db ] - return ResponseMessage.success(value=response_data) + return ResponseMessage.success(value=response_data, code=CommonCode.SUCCESS_GET_API_KEY) @router.get( @@ -83,7 +83,7 @@ def get_api_key_by_service_name( created_at=api_key_in_db.created_at, updated_at=api_key_in_db.updated_at, ) - return ResponseMessage.success(value=response_data) + return ResponseMessage.success(value=response_data, code=CommonCode.SUCCESS_GET_API_KEY) @router.put( @@ -110,7 +110,7 @@ def update_api_key( updated_at=updated_api_key.updated_at, ) - return ResponseMessage.success(value=response_data) + return ResponseMessage.success(value=response_data, code=CommonCode.SUCCESS_UPDATE_API_KEY) @router.delete( @@ -124,4 +124,4 @@ def delete_api_key(serviceName: LLMServiceEnum, service: APIKeyService = api_key - **service_name**: 삭제할 서비스의 이름 """ service.delete_api_key(serviceName.value) - return ResponseMessage.success() + return ResponseMessage.success(code=CommonCode.SUCCESS_DELETE_API_KEY) diff --git a/app/core/enum/db_key_prefix_name.py b/app/core/enum/db_key_prefix_name.py index 5eea582..859ed2f 100644 --- a/app/core/enum/db_key_prefix_name.py +++ b/app/core/enum/db_key_prefix_name.py @@ -1,7 +1,10 @@ # 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 + driver = "DRIVER" + api_key = "API-KEY" diff --git a/app/core/status.py b/app/core/status.py index 1dc25c5..6da4e25 100644 --- a/app/core/status.py +++ b/app/core/status.py @@ -30,6 +30,9 @@ class CommonCode(Enum): """ KEY 성공 코드 - 22xx """ + SUCCESS_DELETE_API_KEY = (status.HTTP_204_NO_CONTENT, "2200", "API KEY가 성공적으로 삭제되었습니다.") + SUCCESS_UPDATE_API_KEY = (status.HTTP_200_OK, "2201", "API KEY가 성공적으로 수정되었습니다.") + SUCCESS_GET_API_KEY = (status.HTTP_200_OK, "2202", "API KEY 정보를 성공적으로 조회했습니다.") """ AI CHAT, DB 성공 코드 - 23xx """ SUCCESS_AI_CHAT_CREATE = (status.HTTP_200_OK, "2300", "새로운 채팅 탭을 생성하였습니다.") diff --git a/app/db/init_db.py b/app/db/init_db.py index 8c5474a..2a83093 100644 --- a/app/db/init_db.py +++ b/app/db/init_db.py @@ -1,26 +1,38 @@ # db/init_db.py -import sqlite3 import logging +import sqlite3 from app.core.utils import get_db_path + def _synchronize_table(cursor, table_name: str, target_columns: dict): """ 테이블 스키마를 확인하고, 코드와 다를 경우 테이블을 재생성하여 동기화합니다. """ try: + # 외래 키 제약 조건 비활성화 + cursor.execute("PRAGMA foreign_keys=off;") 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()} + target_schema_simple = { + name: definition.split()[0].upper() + for name, definition in target_columns.items() + if not name.startswith("FOREIGN KEY") + } if current_columns == target_schema_simple: + # 외래 키 제약 조건 다시 활성화 + cursor.execute("PRAGMA foreign_keys=on;") return - logging.warning(f"'{table_name}' 테이블의 스키마 변경을 감지했습니다. 마이그레이션을 시작합니다. (데이터 손실 위험)") + logging.warning( + f"'{table_name}' 테이블의 스키마 변경을 감지했습니다. 마이그레이션을 시작합니다. (데이터 손실 위험)" + ) temp_table_name = f"{table_name}_temp_old" + cursor.execute(f"DROP TABLE IF EXISTS {temp_table_name}") # DROP 먼저 실행 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()]) @@ -31,7 +43,9 @@ def _synchronize_table(cursor, table_name: str, target_columns: dict): 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}") + 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}") @@ -40,6 +54,9 @@ def _synchronize_table(cursor, table_name: str, target_columns: dict): except sqlite3.Error as e: logging.error(f"'{table_name}' 테이블 마이그레이션 중 오류 발생: {e}") raise e + finally: + # 외래 키 제약 조건 다시 활성화 + cursor.execute("PRAGMA foreign_keys=on;") def initialize_database(): @@ -64,9 +81,12 @@ def initialize_database(): "password": "VARCHAR(128)", "view_name": "VARCHAR(64)", "created_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP", - "updated_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()])})") + 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) cursor.execute( @@ -83,9 +103,10 @@ def initialize_database(): "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" + "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()])})") + create_sql = f"CREATE TABLE IF NOT EXISTS ai_credential ({', '.join([f'{k} {v}' for k, v in ai_credential_cols.items()])})" + cursor.execute(create_sql) _synchronize_table(cursor, "ai_credential", ai_credential_cols) cursor.execute( @@ -101,9 +122,12 @@ def initialize_database(): "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" + "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()])})") + create_sql = ( + f"CREATE TABLE IF NOT EXISTS chat_tab ({', '.join([f'{k} {v}' for k, v in chat_tab_cols.items()])})" + ) + cursor.execute(create_sql) _synchronize_table(cursor, "chat_tab", chat_tab_cols) cursor.execute( """ @@ -121,12 +145,15 @@ def initialize_database(): "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)" + "FOREIGN KEY (chat_tab_id)": "REFERENCES chat_tab(id) ON DELETE CASCADE", } - 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")}) + create_sql = ( + f"CREATE TABLE IF NOT EXISTS chat_message ({', '.join([f'{k} {v}' for k, v in chat_message_cols.items()])})" + ) + cursor.execute(create_sql) + _synchronize_table( + cursor, "chat_message", {k: v for k, v in chat_message_cols.items() if not k.startswith("FOREIGN KEY")} + ) cursor.execute( """ @@ -145,12 +172,13 @@ def initialize_database(): "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)" + "FOREIGN KEY (chat_message_id)": "REFERENCES chat_message(id) ON DELETE CASCADE", } - 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")}) + create_sql = f"CREATE TABLE IF NOT EXISTS query_history ({', '.join([f'{k} {v}' for k, v in query_history_cols.items()])})" + cursor.execute(create_sql) + _synchronize_table( + cursor, "query_history", {k: v for k, v in query_history_cols.items() if not k.startswith("FOREIGN KEY")} + ) cursor.execute( """ @@ -160,252 +188,219 @@ def initialize_database(): """ ) - # database_annotation 테이블 생성 - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS database_annotation ( - id VARCHAR(64) PRIMARY KEY NOT NULL, - db_profile_id VARCHAR(64) NOT NULL, - database_name VARCHAR(255) NOT NULL, - description TEXT, - created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, - updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, - FOREIGN KEY (db_profile_id) REFERENCES db_profile(id) ON DELETE CASCADE - ); - """ + # --- database_annotation 테이블 처리 --- + database_annotation_cols = { + "id": "VARCHAR(64) PRIMARY KEY NOT NULL", + "db_profile_id": "VARCHAR(64) NOT NULL", + "database_name": "VARCHAR(255) NOT NULL", + "description": "TEXT", + "created_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP", + "updated_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP", + "FOREIGN KEY (db_profile_id)": "REFERENCES db_profile(id) ON DELETE CASCADE", + } + create_sql = f"CREATE TABLE IF NOT EXISTS database_annotation ({', '.join([f'{k} {v}' for k, v in database_annotation_cols.items()])})" + cursor.execute(create_sql) + _synchronize_table( + cursor, + "database_annotation", + {k: v for k, v in database_annotation_cols.items() if not k.startswith("FOREIGN KEY")}, ) - # database_annotation 테이블의 updated_at을 자동으로 업데이트하는 트리거 cursor.execute( """ CREATE TRIGGER IF NOT EXISTS update_database_annotation_updated_at - BEFORE UPDATE ON database_annotation - FOR EACH ROW - BEGIN - UPDATE database_annotation SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; - END; + BEFORE UPDATE ON database_annotation FOR EACH ROW + BEGIN UPDATE database_annotation SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; END; """ ) - # table_annotation 테이블 생성 - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS table_annotation ( - id VARCHAR(64) PRIMARY KEY NOT NULL, - database_annotation_id VARCHAR(64) NOT NULL, - table_name VARCHAR(255) NOT NULL, - description TEXT, - created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, - updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, - FOREIGN KEY (database_annotation_id) REFERENCES database_annotation(id) ON DELETE CASCADE - ); - """ + # --- table_annotation 테이블 처리 --- + table_annotation_cols = { + "id": "VARCHAR(64) PRIMARY KEY NOT NULL", + "database_annotation_id": "VARCHAR(64) NOT NULL", + "table_name": "VARCHAR(255) NOT NULL", + "description": "TEXT", + "created_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP", + "updated_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP", + "FOREIGN KEY (database_annotation_id)": "REFERENCES database_annotation(id) ON DELETE CASCADE", + } + create_sql = f"CREATE TABLE IF NOT EXISTS table_annotation ({', '.join([f'{k} {v}' for k, v in table_annotation_cols.items()])})" + cursor.execute(create_sql) + _synchronize_table( + cursor, + "table_annotation", + {k: v for k, v in table_annotation_cols.items() if not k.startswith("FOREIGN KEY")}, ) - # table_annotation 테이블의 updated_at을 자동으로 업데이트하는 트리거 cursor.execute( """ CREATE TRIGGER IF NOT EXISTS update_table_annotation_updated_at - BEFORE UPDATE ON table_annotation - FOR EACH ROW - BEGIN - UPDATE table_annotation SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; - END; + BEFORE UPDATE ON table_annotation FOR EACH ROW + BEGIN UPDATE table_annotation SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; END; """ ) - # column_annotation 테이블 생성 (단일 컬럼 스펙 전용) - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS column_annotation ( - id VARCHAR(64) PRIMARY KEY NOT NULL, - table_annotation_id VARCHAR(64) NOT NULL, - column_name VARCHAR(255) NOT NULL, - -- 데이터 타입 (원본 DB의 타입 문자열을 그대로 저장; 예: BIGINT, TEXT, TIMESTAMP) - data_type VARCHAR(64), - -- NULL 허용 여부 (1:true, 0:false) - is_nullable INTEGER NOT NULL DEFAULT 1, - -- 기본값(리터럴 또는 표현식; 문자열 형태로 저장) - default_value TEXT, - -- 단일 컬럼 기준 CHECK 제약 표현(예: "value > 0") - check_expression TEXT, - -- 컬럼 순서 - ordinal_position INTEGER, - -- 설명 - description TEXT, - created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, - updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, - FOREIGN KEY (table_annotation_id) REFERENCES table_annotation(id) ON DELETE CASCADE - ); - """ + # --- column_annotation 테이블 처리 --- + column_annotation_cols = { + "id": "VARCHAR(64) PRIMARY KEY NOT NULL", + "table_annotation_id": "VARCHAR(64) NOT NULL", + "column_name": "VARCHAR(255) NOT NULL", + "data_type": "VARCHAR(64)", + "is_nullable": "INTEGER NOT NULL DEFAULT 1", + "default_value": "TEXT", + "check_expression": "TEXT", + "ordinal_position": "INTEGER", + "description": "TEXT", + "created_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP", + "updated_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP", + "FOREIGN KEY (table_annotation_id)": "REFERENCES table_annotation(id) ON DELETE CASCADE", + } + create_sql = f"CREATE TABLE IF NOT EXISTS column_annotation ({', '.join([f'{k} {v}' for k, v in column_annotation_cols.items()])})" + cursor.execute(create_sql) + _synchronize_table( + cursor, + "column_annotation", + {k: v for k, v in column_annotation_cols.items() if not k.startswith("FOREIGN KEY")}, ) - # column_annotation 테이블의 updated_at을 자동으로 업데이트하는 트리거 cursor.execute( """ CREATE TRIGGER IF NOT EXISTS update_column_annotation_updated_at - BEFORE UPDATE ON column_annotation - FOR EACH ROW - BEGIN - UPDATE column_annotation SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; - END; + BEFORE UPDATE ON column_annotation FOR EACH ROW + BEGIN UPDATE column_annotation SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; END; """ ) - # table_relationship 테이블 생성 - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS table_relationship ( - id VARCHAR(64) PRIMARY KEY NOT NULL, - database_annotation_id VARCHAR(64) NOT NULL, - from_table_id VARCHAR(64) NOT NULL, - to_table_id VARCHAR(64) NOT NULL, - relationship_type VARCHAR(32) NOT NULL, - description TEXT, - created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, - updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, - FOREIGN KEY (database_annotation_id) REFERENCES database_annotation(id) ON DELETE CASCADE, - FOREIGN KEY (from_table_id) REFERENCES table_annotation(id) ON DELETE CASCADE, - FOREIGN KEY (to_table_id) REFERENCES table_annotation(id) ON DELETE CASCADE - ); - """ + # --- table_relationship 테이블 처리 --- + table_relationship_cols = { + "id": "VARCHAR(64) PRIMARY KEY NOT NULL", + "database_annotation_id": "VARCHAR(64) NOT NULL", + "from_table_id": "VARCHAR(64) NOT NULL", + "to_table_id": "VARCHAR(64) NOT NULL", + "relationship_type": "VARCHAR(32) NOT NULL", + "description": "TEXT", + "created_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP", + "updated_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP", + "FOREIGN KEY (database_annotation_id)": "REFERENCES database_annotation(id) ON DELETE CASCADE", + "FOREIGN KEY (from_table_id)": "REFERENCES table_annotation(id) ON DELETE CASCADE", + "FOREIGN KEY (to_table_id)": "REFERENCES table_annotation(id) ON DELETE CASCADE", + } + create_sql = f"CREATE TABLE IF NOT EXISTS table_relationship ({', '.join([f'{k} {v}' for k, v in table_relationship_cols.items()])})" + cursor.execute(create_sql) + _synchronize_table( + cursor, + "table_relationship", + {k: v for k, v in table_relationship_cols.items() if not k.startswith("FOREIGN KEY")}, ) - # table_relationship 테이블의 updated_at을 자동으로 업데이트하는 트리거 cursor.execute( """ CREATE TRIGGER IF NOT EXISTS update_table_relationship_updated_at - BEFORE UPDATE ON table_relationship - FOR EACH ROW - BEGIN - UPDATE table_relationship SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; - END; + BEFORE UPDATE ON table_relationship FOR EACH ROW + BEGIN UPDATE table_relationship SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; END; """ ) - # --------------------------------------------------------------------- - # 복합 제약(Primary/Unique/ForeignKey/Check) 메타데이터 테이블 생성 - # - 여러 컬럼을 묶는 제약을 '그룹' 단위로 관리 - # - UI 배지/목록은 이 테이블들에서 파생 계산 - # --------------------------------------------------------------------- - - # table_constraint 테이블 생성 (제약 그룹 본체) - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS table_constraint ( - id VARCHAR(64) PRIMARY KEY NOT NULL, - table_annotation_id VARCHAR(64) NOT NULL, - -- PRIMARY_KEY | UNIQUE | FOREIGN_KEY | CHECK - constraint_type VARCHAR(16) NOT NULL, - -- DB 제약명(선택) - name VARCHAR(255), - -- CHECK 제약식 등 (FK/PK/UNIQUE에는 NULL 가능) - expression TEXT, - -- FOREIGN KEY 전용: 참조 테이블명 - ref_table VARCHAR(255), - -- FOREIGN KEY 전용: 액션 - on_update_action VARCHAR(16), - on_delete_action VARCHAR(16), - created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, - updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, - FOREIGN KEY (table_annotation_id) REFERENCES table_annotation(id) ON DELETE CASCADE - ); - """ + # --- table_constraint 테이블 처리 --- + table_constraint_cols = { + "id": "VARCHAR(64) PRIMARY KEY NOT NULL", + "table_annotation_id": "VARCHAR(64) NOT NULL", + "constraint_type": "VARCHAR(16) NOT NULL", + "name": "VARCHAR(255)", + "expression": "TEXT", + "ref_table": "VARCHAR(255)", + "on_update_action": "VARCHAR(16)", + "on_delete_action": "VARCHAR(16)", + "created_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP", + "updated_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP", + "FOREIGN KEY (table_annotation_id)": "REFERENCES table_annotation(id) ON DELETE CASCADE", + } + create_sql = f"CREATE TABLE IF NOT EXISTS table_constraint ({', '.join([f'{k} {v}' for k, v in table_constraint_cols.items()])})" + cursor.execute(create_sql) + _synchronize_table( + cursor, + "table_constraint", + {k: v for k, v in table_constraint_cols.items() if not k.startswith("FOREIGN KEY")}, ) - # table_constraint 테이블의 updated_at을 자동으로 업데이트하는 트리거 cursor.execute( """ CREATE TRIGGER IF NOT EXISTS update_table_constraint_updated_at - BEFORE UPDATE ON table_constraint - FOR EACH ROW - BEGIN - UPDATE table_constraint SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; - END; + BEFORE UPDATE ON table_constraint FOR EACH ROW + BEGIN UPDATE table_constraint SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; END; """ ) - # constraint_column 테이블 생성 (제약 그룹 ↔ 컬럼 매핑) - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS constraint_column ( - id VARCHAR(64) PRIMARY KEY NOT NULL, - constraint_id VARCHAR(64) NOT NULL, - column_annotation_id VARCHAR(64) NOT NULL, - -- 복합 제약 내 컬럼 순서(1, 2, 3, ...) - position INTEGER, - -- FOREIGN KEY 전용: 참조 테이블의 대응 컬럼명 (복합 FK 매핑) - referenced_column_name VARCHAR(255), - created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, - updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, - FOREIGN KEY (constraint_id) REFERENCES table_constraint(id) ON DELETE CASCADE, - FOREIGN KEY (column_annotation_id) REFERENCES column_annotation(id) ON DELETE CASCADE - ); - """ + # --- constraint_column 테이블 처리 --- + constraint_column_cols = { + "id": "VARCHAR(64) PRIMARY KEY NOT NULL", + "constraint_id": "VARCHAR(64) NOT NULL", + "column_annotation_id": "VARCHAR(64) NOT NULL", + "position": "INTEGER", + "referenced_column_name": "VARCHAR(255)", + "created_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP", + "updated_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP", + "FOREIGN KEY (constraint_id)": "REFERENCES table_constraint(id) ON DELETE CASCADE", + "FOREIGN KEY (column_annotation_id)": "REFERENCES column_annotation(id) ON DELETE CASCADE", + } + create_sql = f"CREATE TABLE IF NOT EXISTS constraint_column ({', '.join([f'{k} {v}' for k, v in constraint_column_cols.items()])})" + cursor.execute(create_sql) + _synchronize_table( + cursor, + "constraint_column", + {k: v for k, v in constraint_column_cols.items() if not k.startswith("FOREIGN KEY")}, ) - # constraint_column 테이블의 updated_at을 자동으로 업데이트하는 트리거 cursor.execute( """ CREATE TRIGGER IF NOT EXISTS update_constraint_column_updated_at - BEFORE UPDATE ON constraint_column - FOR EACH ROW - BEGIN - UPDATE constraint_column SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; - END; + BEFORE UPDATE ON constraint_column FOR EACH ROW + BEGIN UPDATE constraint_column SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; END; """ ) - # --------------------------------------------------------------------- - # 인덱스(복합 포함) 메타데이터 테이블 생성 - # - DB 인덱스명을 보존하고, 컬럼 순서를 기록 - # --------------------------------------------------------------------- - - # index_annotation 테이블 생성 (인덱스 그룹 본체) - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS index_annotation ( - id VARCHAR(64) PRIMARY KEY NOT NULL, - table_annotation_id VARCHAR(64) NOT NULL, - name VARCHAR(255), -- DB 인덱스명(선택) - is_unique INTEGER NOT NULL DEFAULT 0, - created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, - updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, - FOREIGN KEY (table_annotation_id) REFERENCES table_annotation(id) ON DELETE CASCADE - ); - """ + # --- index_annotation 테이블 처리 --- + index_annotation_cols = { + "id": "VARCHAR(64) PRIMARY KEY NOT NULL", + "table_annotation_id": "VARCHAR(64) NOT NULL", + "name": "VARCHAR(255)", + "is_unique": "INTEGER NOT NULL DEFAULT 0", + "created_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP", + "updated_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP", + "FOREIGN KEY (table_annotation_id)": "REFERENCES table_annotation(id) ON DELETE CASCADE", + } + create_sql = f"CREATE TABLE IF NOT EXISTS index_annotation ({', '.join([f'{k} {v}' for k, v in index_annotation_cols.items()])})" + cursor.execute(create_sql) + _synchronize_table( + cursor, + "index_annotation", + {k: v for k, v in index_annotation_cols.items() if not k.startswith("FOREIGN KEY")}, ) - # index_annotation 테이블의 updated_at을 자동으로 업데이트하는 트리거 cursor.execute( """ CREATE TRIGGER IF NOT EXISTS update_index_annotation_updated_at - BEFORE UPDATE ON index_annotation - FOR EACH ROW - BEGIN - UPDATE index_annotation SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; - END; + BEFORE UPDATE ON index_annotation FOR EACH ROW + BEGIN UPDATE index_annotation SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; END; """ ) - # index_column 테이블 생성 (인덱스 그룹 ↔ 컬럼 매핑) - cursor.execute( - """ - CREATE TABLE IF NOT EXISTS index_column ( - id VARCHAR(64) PRIMARY KEY NOT NULL, - index_id VARCHAR(64) NOT NULL, - column_annotation_id VARCHAR(64) NOT NULL, - -- 인덱스 내 컬럼 순서(1, 2, 3, ...) - position INTEGER, - created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, - updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, - FOREIGN KEY (index_id) REFERENCES index_annotation(id) ON DELETE CASCADE, - FOREIGN KEY (column_annotation_id) REFERENCES column_annotation(id) ON DELETE CASCADE - ); - """ + # --- index_column 테이블 처리 --- + index_column_cols = { + "id": "VARCHAR(64) PRIMARY KEY NOT NULL", + "index_id": "VARCHAR(64) NOT NULL", + "column_annotation_id": "VARCHAR(64) NOT NULL", + "position": "INTEGER", + "created_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP", + "updated_at": "DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP", + "FOREIGN KEY (index_id)": "REFERENCES index_annotation(id) ON DELETE CASCADE", + "FOREIGN KEY (column_annotation_id)": "REFERENCES column_annotation(id) ON DELETE CASCADE", + } + create_sql = ( + f"CREATE TABLE IF NOT EXISTS index_column ({', '.join([f'{k} {v}' for k, v in index_column_cols.items()])})" + ) + cursor.execute(create_sql) + _synchronize_table( + cursor, "index_column", {k: v for k, v in index_column_cols.items() if not k.startswith("FOREIGN KEY")} ) - # index_column 테이블의 updated_at을 자동으로 업데이트하는 트리거 cursor.execute( """ CREATE TRIGGER IF NOT EXISTS update_index_column_updated_at - BEFORE UPDATE ON index_column - FOR EACH ROW - BEGIN - UPDATE index_column SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; - END; + BEFORE UPDATE ON index_column FOR EACH ROW + BEGIN UPDATE index_column SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id; END; """ ) diff --git a/app/services/api_key_service.py b/app/services/api_key_service.py index 7e43cb9..36b7492 100644 --- a/app/services/api_key_service.py +++ b/app/services/api_key_service.py @@ -2,6 +2,7 @@ from fastapi import Depends +from app.core.enum.db_key_prefix_name import DBSaveIdEnum from app.core.exceptions import APIException from app.core.security import AES256 from app.core.status import CommonCode @@ -23,7 +24,7 @@ def store_api_key(self, api_key_data: APIKeyCreate) -> APIKeyInDB: api_key_data.validate_with_service() try: encrypted_key = AES256.encrypt(api_key_data.api_key) - new_id = generate_prefixed_uuid("QGENIE") + new_id = generate_prefixed_uuid(DBSaveIdEnum.api_key.value) created_row = self.repository.create_api_key( new_id=new_id,