백엔드 API 구현 — 데이터베이스와 HTTP를 연결하다

지금까지 배운 것들을 이어붙이는 시간

1장에서 API가 무엇인지 배웠습니다.
2장에서 웹 서비스가 어떤 구조로 동작하는지 배웠습니다.
5장에서 데이터베이스를 설계했습니다.
6장에서 SQL로 데이터를 다루는 법을 배웠습니다.

이제 이것들을 하나로 연결할 차례입니다.

클라이언트 (브라우저 / 앱)
       ↕  HTTP 요청 / 응답
   백엔드 서버  ← [이 장에서 만드는 것]
       ↕  SQL
   데이터베이스

REST API 설계 원칙

백엔드가 클라이언트에게 제공하는 API를 어떻게 설계해야 할까요?

가장 널리 쓰이는 방식이 REST(Representational State Transfer)입니다.

핵심 규칙

1. URL은 명사(자원)를 나타낸다

좋은 예 (명사):         나쁜 예 (동사):
/todos                  /getTodos
/todos/5                /deleteTodo?id=5
/users/1/todos          /getUserTodos?userId=1

2. HTTP 메서드가 행동을 나타낸다

메서드	의미	예시
GET	조회	GET /todos — 목록 조회
POST	생성	POST /todos — 새 항목 생성
PUT	전체 수정	PUT /todos/5 — 5번 항목 전체 교체
PATCH	일부 수정	PATCH /todos/5 — 5번 항목 일부 수정
DELETE	삭제	DELETE /todos/5 — 5번 항목 삭제

3. 응답은 JSON으로

{
  "id": 5,
  "title": "운동하기",
  "is_done": false,
  "created_at": "2026-05-21T10:00:00"
}

Todo 서비스 API 설계 예시

메서드	URL	설명
GET	/todos	전체 목록 조회
GET	/todos/5	5번 항목 조회
POST	/todos	새 항목 생성
PATCH	/todos/5	5번 항목 수정
DELETE	/todos/5	5번 항목 삭제

HTTP 상태 코드

서버는 응답할 때 상태 코드로 결과를 알립니다.

코드	의미	상황
200 OK	성공	GET, PATCH, DELETE 성공
201 Created	생성 성공	POST 성공
400 Bad Request	잘못된 요청	필수 값 누락, 형식 오류
401 Unauthorized	인증 필요	로그인 안 됨
403 Forbidden	권한 없음	다른 사람 데이터에 접근
404 Not Found	없는 자원	존재하지 않는 ID
500 Internal Server Error	서버 오류	예외 발생, DB 오류

성공: 2xx
클라이언트 실수: 4xx
서버 실수: 5xx

Python으로 백엔드 만들기 — FastAPI

이 강의에서는 FastAPI를 사용합니다.
간결하고, 자동으로 API 문서를 생성해 주는 Python 프레임워크입니다.

pip install fastapi uvicorn

가장 단순한 서버

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def root():
    return {"message": "Hello, World!"}

uvicorn main:app --reload
# http://localhost:8000 에서 동작
# http://localhost:8000/docs 에서 API 문서 자동 생성

데이터베이스 연결 — SQLite + sqlite3

Python 표준 라이브러리 sqlite3로 데이터베이스에 연결합니다.

import sqlite3

def get_db():
    conn = sqlite3.connect("app.db")
    conn.row_factory = sqlite3.Row  # 딕셔너리처럼 접근 가능
    return conn

테이블 초기화

def init_db():
    conn = get_db()
    conn.execute("""
        CREATE TABLE IF NOT EXISTS todos (
            id         INTEGER PRIMARY KEY AUTOINCREMENT,
            title      TEXT    NOT NULL,
            is_done    INTEGER DEFAULT 0,
            created_at TEXT    DEFAULT (datetime('now'))
        )
    """)
    conn.commit()
    conn.close()

CRUD API 구현

CRUD = Create / Read / Update / Delete

GET /todos — 목록 조회

from fastapi import FastAPI
import sqlite3

app = FastAPI()

@app.get("/todos")
def get_todos():
    conn = sqlite3.connect("app.db")
    conn.row_factory = sqlite3.Row
    rows = conn.execute("SELECT * FROM todos ORDER BY created_at DESC").fetchall()
    conn.close()
    return [dict(row) for row in rows]

응답:

[
  {"id": 2, "title": "책 읽기",  "is_done": 0, "created_at": "..."},
  {"id": 1, "title": "운동하기", "is_done": 1, "created_at": "..."}
]

GET /todos/{id} — 단건 조회

from fastapi import HTTPException

@app.get("/todos/{todo_id}")
def get_todo(todo_id: int):
    conn = sqlite3.connect("app.db")
    conn.row_factory = sqlite3.Row
    row = conn.execute("SELECT * FROM todos WHERE id = ?", (todo_id,)).fetchone()
    conn.close()

    if row is None:
        raise HTTPException(status_code=404, detail="Not found")

    return dict(row)

? 자리에 값을 넣는 방식을 파라미터 바인딩이라고 합니다.
SQL 문자열에 값을 직접 삽입하면 SQL 인젝션 취약점이 생깁니다.

POST /todos — 생성

from pydantic import BaseModel

class TodoCreate(BaseModel):
    title: str

@app.post("/todos", status_code=201)
def create_todo(body: TodoCreate):
    conn = sqlite3.connect("app.db")
    cursor = conn.execute(
        "INSERT INTO todos (title) VALUES (?)",
        (body.title,)
    )
    conn.commit()
    new_id = cursor.lastrowid
    conn.close()
    return {"id": new_id, "title": body.title, "is_done": 0}

요청:

POST /todos
{"title": "영어 공부"}

응답:

HTTP 201
{"id": 3, "title": "영어 공부", "is_done": 0}

PATCH /todos/{id} — 수정

class TodoUpdate(BaseModel):
    title: str | None = None
    is_done: bool | None = None

@app.patch("/todos/{todo_id}")
def update_todo(todo_id: int, body: TodoUpdate):
    conn = sqlite3.connect("app.db")
    conn.row_factory = sqlite3.Row

    row = conn.execute("SELECT * FROM todos WHERE id = ?", (todo_id,)).fetchone()
    if row is None:
        conn.close()
        raise HTTPException(status_code=404, detail="Not found")

    new_title   = body.title   if body.title   is not None else row["title"]
    new_is_done = 1 if body.is_done else (0 if body.is_done is not None else row["is_done"])

    conn.execute(
        "UPDATE todos SET title = ?, is_done = ? WHERE id = ?",
        (new_title, new_is_done, todo_id)
    )
    conn.commit()
    conn.close()
    return {"id": todo_id, "title": new_title, "is_done": bool(new_is_done)}

DELETE /todos/{id} — 삭제

@app.delete("/todos/{todo_id}", status_code=200)
def delete_todo(todo_id: int):
    conn = sqlite3.connect("app.db")

    result = conn.execute("DELETE FROM todos WHERE id = ?", (todo_id,))
    conn.commit()
    conn.close()

    if result.rowcount == 0:
        raise HTTPException(status_code=404, detail="Not found")

    return {"deleted": todo_id}

요청-응답 전체 흐름

클라이언트                      FastAPI 서버                   SQLite DB
    │                               │                              │
    │  POST /todos                  │                              │
    │  {"title": "영어 공부"}  ────▶│                              │
    │                               │  INSERT INTO todos ...  ────▶│
    │                               │                              │
    │                               │  ◀──── lastrowid = 3        │
    │                               │                              │
    │  ◀──── HTTP 201               │                              │
    │  {"id": 3, "title": ...}      │                              │

입력 검증 (Validation)

클라이언트에서 잘못된 값이 올 수 있습니다.
FastAPI + Pydantic은 자동으로 검증하고 400 응답을 돌려줍니다.

from pydantic import BaseModel, Field

class TodoCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)

# title이 비어있거나 200자를 넘으면 자동으로 400 반환

요청:

POST /todos
{"title": ""}

응답:

HTTP 400
{"detail": [{"type": "string_too_short", "loc": ["body", "title"], ...}]}

쿼리 파라미터

URL 뒤에 ?key=value 형태로 전달하는 값입니다.

@app.get("/todos")
def get_todos(is_done: int | None = None, limit: int = 20):
    query = "SELECT * FROM todos"
    params = []

    if is_done is not None:
        query += " WHERE is_done = ?"
        params.append(is_done)

    query += " ORDER BY created_at DESC LIMIT ?"
    params.append(limit)

    conn = sqlite3.connect("app.db")
    conn.row_factory = sqlite3.Row
    rows = conn.execute(query, params).fetchall()
    conn.close()
    return [dict(row) for row in rows]

GET /todos              → 전체 목록 (최대 20개)
GET /todos?is_done=0    → 미완료만
GET /todos?limit=5      → 최대 5개만
GET /todos?is_done=1&limit=3 → 완료된 것 중 3개

전체 파일 구조

project/
├── main.py        ← FastAPI 앱, 라우터
├── database.py    ← DB 연결, 초기화
└── app.db         ← SQLite 데이터베이스 파일 (자동 생성)

# database.py
import sqlite3

DATABASE = "app.db"

def get_db():
    conn = sqlite3.connect(DATABASE)
    conn.row_factory = sqlite3.Row
    return conn

def init_db():
    conn = get_db()
    conn.execute("""
        CREATE TABLE IF NOT EXISTS todos (
            id         INTEGER PRIMARY KEY AUTOINCREMENT,
            title      TEXT    NOT NULL,
            is_done    INTEGER DEFAULT 0,
            created_at TEXT    DEFAULT (datetime('now'))
        )
    """)
    conn.commit()
    conn.close()

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from database import get_db, init_db

app = FastAPI()
init_db()

class TodoCreate(BaseModel):
    title: str = Field(min_length=1, max_length=200)

class TodoUpdate(BaseModel):
    title: str | None = Field(default=None, min_length=1, max_length=200)
    is_done: bool | None = None

@app.get("/todos")
def get_todos():
    conn = get_db()
    rows = conn.execute("SELECT * FROM todos ORDER BY created_at DESC").fetchall()
    conn.close()
    return [dict(row) for row in rows]

@app.post("/todos", status_code=201)
def create_todo(body: TodoCreate):
    conn = get_db()
    cursor = conn.execute("INSERT INTO todos (title) VALUES (?)", (body.title,))
    conn.commit()
    new_id = cursor.lastrowid
    conn.close()
    return {"id": new_id, "title": body.title, "is_done": False}

@app.delete("/todos/{todo_id}")
def delete_todo(todo_id: int):
    conn = get_db()
    result = conn.execute("DELETE FROM todos WHERE id = ?", (todo_id,))
    conn.commit()
    conn.close()
    if result.rowcount == 0:
        raise HTTPException(status_code=404, detail="Not found")
    return {"deleted": todo_id}

SQL 인젝션 — 절대 하지 말 것

# 위험: 사용자 입력을 문자열로 직접 조합
title = request.query_params["title"]
conn.execute(f"SELECT * FROM todos WHERE title = '{title}'")

# 공격자가 title에 이것을 넣으면:
# ' OR '1'='1
# 쿼리: SELECT * FROM todos WHERE title = '' OR '1'='1'
# → 전체 데이터가 노출!

# 안전: 파라미터 바인딩 사용
conn.execute("SELECT * FROM todos WHERE title = ?", (title,))
# sqlite3가 값을 안전하게 처리함

실습 미션

미션 1: 서버 실행

1. main.py와 database.py를 작성하고 서버를 실행하세요.
2. http://localhost:8000/docs 에서 자동 생성된 API 문서를 확인하세요.
3. 문서 페이지에서 직접 API를 호출해 보세요.

미션 2: CRUD 완성

1. GET /todos/{id} 엔드포인트를 추가하세요.
2. PATCH /todos/{id} 엔드포인트를 추가하세요.
3. 각각 404 처리를 포함하세요.

미션 3: 기능 추가

1. GET /todos?is_done=0 처럼 필터링이 가능하게 만드세요.
2. GET /todos?limit=5 처럼 개수 제한이 가능하게 만드세요.
3. 완료된 할 일을 일괄 삭제하는 DELETE /todos/done 엔드포인트를 만드세요.

미션 4 (심화): 다중 사용자

users 테이블을 추가하고, todos에 user_id FK를 연결하세요.
- POST /users — 사용자 생성
- GET /users/{id}/todos — 특정 사용자의 할 일 목록 조회
- POST /users/{id}/todos — 특정 사용자의 할 일 생성

핵심 요약

개념	설명
REST	URL=자원, HTTP 메서드=행동, JSON=데이터
상태 코드	2xx 성공, 4xx 클라이언트 오류, 5xx 서버 오류
파라미터 바인딩	? 플레이스홀더 — SQL 인젝션 방지
Pydantic	요청 Body 자동 검증
FastAPI	라우팅 + 검증 + 자동 문서화

DB에 값을 넣을 때는 항상 파라미터 바인딩(?)을 사용하세요.
문자열 포매팅(f"...{value}")으로 SQL을 만드는 건 보안 취약점입니다.