전체 코드는 아래에
"""
- 표준 라이브러리 http.server만 사용
- DB는 SQLite + raw SQL
엔드포인트:
GET /users
POST /users
GET /users/<id>
PUT /users/<id>
DELETE /users/<id>
"""
import json
import re
import sqlite3
from http.server import BaseHTTPRequestHandler, HTTPServer
DB_PATH = "users.db"
def init_db():
conn = sqlite3.connect(DB_PATH)
conn.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT NOT NULL
)
""")
conn.commit()
conn.close()
def get_conn():
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
return conn
def db_list_users():
conn = get_conn()
rows = conn.execute("SELECT id, name, email FROM users").fetchall()
conn.close()
return [dict(r) for r in rows]
RE_COLLECTION = re.compile(r"^/users/?$")
RE_ITEM = re.compile(r"^/users/(\d+)/?$")
class CRUDHandler(BaseHTTPRequestHandler):
def _send_json(self, status, payload):
...
def _read_json_body(self):
...
def do_get(self):
if RE_COLLECTION.match(self.path):
self._send_json(200, db_list_users())
return
def run(port=8000):
init_db()
server = HTTPServer(("127.0.0.1", port), CRUDHandler)
print(f"serving on <http://127.0.0.1>:{port})
server.serve_forever()
if __name__ == "__main__":
run()
BastHTTPRequestHandler를 상속하는 것은 행위(behavorial)패턴 중 템플릿 메서드 (Template Method) 패턴입니다. 상위 클래스 (BaseHTTPRequestHandler)는 골격이 되는 코드를 고정적으로 갖고 있어요: handle_one_request() — 요청 라인 파싱 → 헤더 파싱 → 메서드 판별 → do_<메서드> 호출
반면 선택적으로 재정의할 수 있는 hook 메서드인 do_*는 저희가 재정의하는 것입니다.
참고로 Template Method 패턴에서는 상위 클래스가 골격 코드는 갖고 있으면서, 하위 클래스가 무조건 재정의 해야되는 추상 메서드 (abstract method)와 선택적으로 재정의할 수 있는 hook 메서드가 있죠.
do_GET은 재정의 즉 override를 안 해도 에러가 안 나요. 대신 상위 클래스에 기본 동작 (501 응답)이 이미 있어서 그게 실행됩니다. 훅 메서드라는 거에요.
참고로 파이썬은 자바의 final 같은 문법이 없기에 골격 코드를 하위 클래스가 override하는 것을 강제적으로 막지는 못하나, 나름의 신호를 주는 방식(관례)들이 있습니다.
무엇보다 중요한 이것은 Inversion of Control 패턴입니다.
Don’t call us, we’ll call you
HTTPServer가 알맞게 CRUDHandler의 do_* 메서드들을 호출하는 것이지, 그 반대 방향이 아닙니다. 이건 Spring의 본질과 같은 맥락의 내용이기 때문에 중요하게 볼 필요가 있습니다.
전체 코드입니다
"""
프레임워크 없이 만든 사용자(User) CRUD 서버
- 표준 라이브러리 http.server 만 사용 (Django/Flask 없음)
- DB는 SQLite + 생(raw) SQL
목적: Django가 '대신 해주던' 것들을 직접 짜보면서,
프레임워크가 없애주는 반복 작업이 무엇인지 체감한다.
엔드포인트:
GET /users 전체 목록
POST /users 생성 (body: {"name": ..., "email": ...})
GET /users/<id> 단건 조회
PUT /users/<id> 수정 (body: {"name": ..., "email": ...})
DELETE /users/<id> 삭제
"""
import json
import re
import sqlite3
from http.server import BaseHTTPRequestHandler, HTTPServer
DB_PATH = "users.db"
# ─────────────────────────────────────────────────────────────
# 0. DB 초기화
# [Django라면] models.py에 class User(models.Model) 만 쓰면
# makemigrations/migrate 가 이 CREATE TABLE 을 자동 생성.
# 여기선 스키마를 손으로 쓴다.
# ─────────────────────────────────────────────────────────────
def init_db():
conn = sqlite3.connect(DB_PATH)
conn.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT NOT NULL
)
""")
conn.commit()
conn.close()
def get_conn():
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row # row를 dict처럼 접근 가능하게
return conn
# ─────────────────────────────────────────────────────────────
# 1. DB 접근 계층 (생 SQL)
# [Django라면] User.objects.all() / .get(id=..) / .create(..)
# ORM이 이 SQL 문자열 + row→객체 변환을 전부 없애준다.
# 여기선 SQL을 손으로 쓰고, ? 바인딩으로 인젝션도 직접 막는다.
# ─────────────────────────────────────────────────────────────
def db_list_users():
conn = get_conn()
rows = conn.execute("SELECT id, name, email FROM users").fetchall()
conn.close()
return [dict(r) for r in rows] # ← 손으로 dict 변환 (ORM이 대신하던 일)
def db_get_user(user_id):
conn = get_conn()
row = conn.execute(
"SELECT id, name, email FROM users WHERE id = ?", (user_id,)
).fetchone()
conn.close()
return dict(row) if row else None
def db_create_user(name, email):
conn = get_conn()
cur = conn.execute(
"INSERT INTO users (name, email) VALUES (?, ?)", (name, email)
)
conn.commit()
new_id = cur.lastrowid
conn.close()
return db_get_user(new_id)
def db_update_user(user_id, name, email):
conn = get_conn()
cur = conn.execute(
"UPDATE users SET name = ?, email = ? WHERE id = ?",
(name, email, user_id),
)
conn.commit()
changed = cur.rowcount # 0이면 그런 id 없었다는 뜻
conn.close()
return db_get_user(user_id) if changed else None
def db_delete_user(user_id):
conn = get_conn()
cur = conn.execute("DELETE FROM users WHERE id = ?", (user_id,))
conn.commit()
changed = cur.rowcount
conn.close()
return changed > 0
# ─────────────────────────────────────────────────────────────
# 2. 요청 핸들러
# [Django라면] urls.py의 path() + views.py 함수/클래스가
# 라우팅과 요청/응답 처리를 담당. 여기선 do_GET/do_POST... 안에서
# 경로를 정규식으로 직접 매칭하고, JSON 파싱/직렬화도 손으로 한다.
# ─────────────────────────────────────────────────────────────
# URL 패턴을 직접 정의. [Django라면] path("users/<int:id>/", ...) 한 줄.
RE_COLLECTION = re.compile(r"^/users/?$") # /users
RE_ITEM = re.compile(r"^/users/(\d+)/?$") # /users/<id>
class CRUDHandler(BaseHTTPRequestHandler):
# ── 공통 헬퍼 ────────────────────────────────────────────
# [Django라면] JsonResponse(obj) 한 줄이면 끝.
# 여기선 상태코드/헤더/직렬화를 매번 직접.
def _send_json(self, status, payload):
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
# [Django라면] request.body 파싱 + json 로딩을 미들웨어/DRF가 처리.
# 여기선 Content-Length 읽어서 바이트 읽고 직접 json.loads.
def _read_json_body(self):
length = int(self.headers.get("Content-Length", 0))
if length == 0:
return {}
raw = self.rfile.read(length)
try:
return json.loads(raw.decode("utf-8"))
except json.JSONDecodeError:
return None # 잘못된 JSON
# ── 라우팅: 메서드별로 경로를 직접 분기 ──────────────────
# [Django라면] 이 전부가 urls.py의 urlpatterns 리스트로 대체됨.
def do_GET(self):
if RE_COLLECTION.match(self.path):
self._send_json(200, db_list_users())
return
m = RE_ITEM.match(self.path)
if m:
user = db_get_user(int(m.group(1))) # ← URL에서 id 직접 추출
if user:
self._send_json(200, user)
else:
self._send_json(404, {"error": "user not found"})
return
self._send_json(404, {"error": "no such route"})
def do_POST(self):
if not RE_COLLECTION.match(self.path):
self._send_json(404, {"error": "no such route"})
return
data = self._read_json_body()
# [Django라면] Form/Serializer가 검증을 대신. 여기선 손으로 체크.
if data is None:
self._send_json(400, {"error": "invalid JSON"})
return
if "name" not in data or "email" not in data:
self._send_json(400, {"error": "name and email required"})
return
user = db_create_user(data["name"], data["email"])
self._send_json(201, user)
def do_PUT(self):
m = RE_ITEM.match(self.path)
if not m:
self._send_json(404, {"error": "no such route"})
return
data = self._read_json_body()
if data is None:
self._send_json(400, {"error": "invalid JSON"})
return
if "name" not in data or "email" not in data:
self._send_json(400, {"error": "name and email required"})
return
user = db_update_user(int(m.group(1)), data["name"], data["email"])
if user:
self._send_json(200, user)
else:
self._send_json(404, {"error": "user not found"})
def do_DELETE(self):
m = RE_ITEM.match(self.path)
if not m:
self._send_json(404, {"error": "no such route"})
return
if db_delete_user(int(m.group(1))):
self._send_json(204, {})
else:
self._send_json(404, {"error": "user not found"})
# 로그를 조용히 (기본은 매 요청 stderr에 찍음)
def log_message(self, *args):
pass
def run(port=8000):
init_db()
server = HTTPServer(("127.0.0.1", port), CRUDHandler)
print(f"serving on <http://127.0.0.1>:{port}")
server.serve_forever()
if __name__ == "__main__":
run()