88 lines
7.3 KiB
Python
88 lines
7.3 KiB
Python
import sqlite3 # SQLite 데이터베이스를 읽고 쓰기 위한 모듈을 가져온다.
|
|
from pathlib import Path # 데이터베이스 파일 경로를 계산하기 위한 모듈을 가져온다.
|
|
|
|
from fastapi import FastAPI, HTTPException # FastAPI 앱과 예외 응답 도구를 가져온다.
|
|
from fastapi.middleware.cors import CORSMiddleware # 프론트엔드 연동을 위한 CORS 미들웨어를 가져온다.
|
|
from pydantic import BaseModel # 요청 본문을 검증하기 위한 기본 모델 클래스를 가져온다.
|
|
|
|
DB_PATH = Path(__file__).resolve().parent / "app.db" # 현재 파일 기준으로 데이터베이스 파일 경로를 정한다.
|
|
app = FastAPI(title="SQLite CRUD API") # FastAPI 애플리케이션 인스턴스를 만든다.
|
|
app.add_middleware(CORSMiddleware, allow_origins=["http://127.0.0.1:5173", "http://localhost:5173"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"]) # Vue 개발 서버에서 API를 호출할 수 있도록 CORS를 허용한다.
|
|
|
|
|
|
class MessagePayload(BaseModel): # 메시지 생성과 수정을 위한 요청 모델을 정의한다.
|
|
content: str # 요청 본문에서 메시지 내용을 문자열로 받는다.
|
|
|
|
|
|
def ensure_database_exists() -> None: # 데이터베이스 파일 존재 여부를 확인하는 함수를 정의한다.
|
|
if not DB_PATH.exists(): # 데이터베이스 파일이 아직 생성되지 않았는지 확인한다.
|
|
raise HTTPException(status_code=500, detail="Database file does not exist. Run backend/init_db.py first.") # 초기화 스크립트 실행이 필요하다는 오류를 반환한다.
|
|
|
|
|
|
def get_connection() -> sqlite3.Connection: # SQLite 연결 객체를 공통으로 만드는 함수를 정의한다.
|
|
ensure_database_exists() # 데이터베이스 파일이 존재하는지 먼저 확인한다.
|
|
connection = sqlite3.connect(DB_PATH) # SQLite 데이터베이스에 연결한다.
|
|
connection.row_factory = sqlite3.Row # 컬럼 이름으로 접근할 수 있도록 Row 팩토리를 지정한다.
|
|
return connection # 설정이 끝난 연결 객체를 반환한다.
|
|
|
|
|
|
def normalize_content(content: str) -> str: # 메시지 내용을 공통 규칙으로 정리하는 함수를 정의한다.
|
|
normalized_content = content.strip() # 앞뒤 공백을 제거해 실제 입력값만 남긴다.
|
|
if not normalized_content: # 공백만 있거나 빈 문자열인 경우를 확인한다.
|
|
raise HTTPException(status_code=400, detail="Content must not be empty") # 빈 메시지는 허용하지 않는다는 오류를 반환한다.
|
|
return normalized_content # 검증을 통과한 메시지 내용을 반환한다.
|
|
|
|
|
|
def serialize_message(row: sqlite3.Row) -> dict[str, int | str]: # SQLite 행 데이터를 JSON 응답용 딕셔너리로 바꾸는 함수를 정의한다.
|
|
return {"id": row["id"], "content": row["content"]} # id와 content만 꺼내서 반환한다.
|
|
|
|
|
|
def read_first_message() -> str: # 데이터베이스에서 첫 번째 메시지를 읽는 함수를 정의한다.
|
|
with get_connection() as connection: # 데이터베이스 연결을 열고 자동으로 닫히게 한다.
|
|
row = connection.execute("SELECT content FROM messages ORDER BY id ASC LIMIT 1").fetchone() # 가장 먼저 저장된 메시지 한 건을 조회한다.
|
|
if row is None: # 조회된 메시지가 없는 경우를 확인한다.
|
|
raise HTTPException(status_code=404, detail="Message not found") # 메시지가 없다는 404 오류를 반환한다.
|
|
return row["content"] # 조회한 메시지 내용을 반환한다.
|
|
|
|
|
|
@app.get("/api/message") # 기존 예제와 호환되도록 첫 번째 메시지를 반환하는 GET 엔드포인트를 유지한다.
|
|
def get_message() -> dict[str, str]: # JSON 응답 형태의 딕셔너리를 반환하는 함수를 정의한다.
|
|
return {"message": read_first_message()} # 데이터베이스에서 읽은 첫 번째 메시지를 JSON으로 반환한다.
|
|
|
|
|
|
@app.get("/api/messages") # 전체 메시지 목록을 반환하는 GET 엔드포인트를 정의한다.
|
|
def list_messages() -> list[dict[str, int | str]]: # 메시지 목록을 JSON 배열 형태로 반환하는 함수를 정의한다.
|
|
with get_connection() as connection: # 데이터베이스 연결을 열고 자동으로 닫히게 한다.
|
|
rows = connection.execute("SELECT id, content FROM messages ORDER BY id ASC").fetchall() # 저장된 메시지를 id 오름차순으로 모두 조회한다.
|
|
return [serialize_message(row) for row in rows] # 조회된 모든 행을 직렬화해 반환한다.
|
|
|
|
|
|
@app.post("/api/messages", status_code=201) # 새 메시지를 저장하는 POST 엔드포인트를 정의한다.
|
|
def create_message(payload: MessagePayload) -> dict[str, int | str]: # 생성된 메시지 정보를 반환하는 함수를 정의한다.
|
|
normalized_content = normalize_content(payload.content) # 요청 본문의 메시지 내용을 공통 규칙으로 정리한다.
|
|
with get_connection() as connection: # 데이터베이스 연결을 열고 자동으로 닫히게 한다.
|
|
cursor = connection.execute("INSERT INTO messages (content) VALUES (?)", (normalized_content,)) # 정리된 메시지 내용을 테이블에 저장한다.
|
|
connection.commit() # INSERT 결과를 데이터베이스에 반영한다.
|
|
return {"id": int(cursor.lastrowid), "content": normalized_content} # 생성된 id와 메시지 내용을 응답으로 반환한다.
|
|
|
|
|
|
@app.put("/api/messages/{message_id}") # 기존 메시지를 수정하는 PUT 엔드포인트를 정의한다.
|
|
def update_message(message_id: int, payload: MessagePayload) -> dict[str, int | str]: # 수정된 메시지 정보를 반환하는 함수를 정의한다.
|
|
normalized_content = normalize_content(payload.content) # 요청 본문의 메시지 내용을 공통 규칙으로 정리한다.
|
|
with get_connection() as connection: # 데이터베이스 연결을 열고 자동으로 닫히게 한다.
|
|
cursor = connection.execute("UPDATE messages SET content = ? WHERE id = ?", (normalized_content, message_id)) # 지정한 id의 메시지 내용을 새 값으로 수정한다.
|
|
connection.commit() # UPDATE 결과를 데이터베이스에 반영한다.
|
|
if cursor.rowcount == 0: # 실제로 수정된 행이 없는 경우를 확인한다.
|
|
raise HTTPException(status_code=404, detail="Message not found") # 없는 메시지라는 404 오류를 반환한다.
|
|
return {"id": message_id, "content": normalized_content} # 수정된 id와 메시지 내용을 응답으로 반환한다.
|
|
|
|
|
|
@app.delete("/api/messages/{message_id}") # 기존 메시지를 삭제하는 DELETE 엔드포인트를 정의한다.
|
|
def delete_message(message_id: int) -> dict[str, bool]: # 삭제 성공 여부를 반환하는 함수를 정의한다.
|
|
with get_connection() as connection: # 데이터베이스 연결을 열고 자동으로 닫히게 한다.
|
|
cursor = connection.execute("DELETE FROM messages WHERE id = ?", (message_id,)) # 지정한 id의 메시지를 테이블에서 삭제한다.
|
|
connection.commit() # DELETE 결과를 데이터베이스에 반영한다.
|
|
if cursor.rowcount == 0: # 실제로 삭제된 행이 없는 경우를 확인한다.
|
|
raise HTTPException(status_code=404, detail="Message not found") # 없는 메시지라는 404 오류를 반환한다.
|
|
return {"success": True} # 삭제가 성공했음을 JSON으로 반환한다.
|