[파이썬 100강] 95강. sqlite3 트랜잭션과 인덱스로 데이터 처리 안정화하기

Last updated on 
[파이썬 100강] 95강. sqlite3 트랜잭션과 인덱스로 데이터 처리 안정화하기

[파이썬 100강] 95강. sqlite3 트랜잭션과 인덱스로 데이터 처리 안정화하기

데이터를 DB에 넣는 코드는 금방 만들 수 있지만, 중간에 실패해도 데이터가 일관되게 남는 코드를 만드는 건 완전히 다른 문제입니다. 오늘은 sqlite3에서 그 차이를 만드는 두 축, 트랜잭션(transaction)과 인덱스(index)를 실전 관점으로 정리합니다. 설명은 짧게 하고 바로 코드로 들어가 보겠습니다.

핵심 개념

  • 트랜잭션은 여러 SQL 작업을 하나의 묶음으로 처리해, 도중 실패 시 전체를 되돌릴 수 있게 해주는 안전장치입니다.
  • COMMIT은 변경사항 확정, ROLLBACK은 마지막 커밋 시점으로 되돌리기입니다.
  • 인덱스는 조회 속도를 높이기 위한 자료구조이며, 읽기를 빠르게 하는 대신 쓰기(INSERT/UPDATE) 비용을 늘릴 수 있습니다.

sqlite3는 파일 기반 DB라서 가볍고 빠르게 시작할 수 있지만, 그렇다고 아무렇게나 써도 된다는 뜻은 아닙니다. 특히 자동화 스크립트나 배치 작업에서 "일부만 저장된 반쪽 데이터"는 나중에 복구 비용을 크게 만듭니다. 트랜잭션을 제대로 쓰면 실패를 의도적으로 만들었을 때도 DB 상태가 깨끗하게 유지됩니다. 그리고 인덱스는 "무조건 많이"가 아니라, 자주 조회하는 조건(WHERE/JOIN/ORDER BY) 중심으로 설계해야 효과가 큽니다. 오늘 예제는 이 두 가지를 함께 묶어, 운영에서 바로 옮겨 쓸 수 있는 형태로 구성합니다.

기본 사용

예제 1) 트랜잭션으로 원자성 보장하기

>>> import sqlite3
>>> conn = sqlite3.connect(":memory:")
>>> cur = conn.cursor()
>>> cur.execute("CREATE TABLE account (id INTEGER PRIMARY KEY, owner TEXT, balance INTEGER)")
<sqlite3.Cursor object at ...>
>>> cur.executemany(
...     "INSERT INTO account(id, owner, balance) VALUES (?, ?, ?)",
...     [(1, "kim", 1000), (2, "lee", 500)]
... )
<sqlite3.Cursor object at ...>
>>> conn.commit()

>>> def transfer(conn, sender_id, receiver_id, amount):
...     with conn:  # 성공 시 commit, 예외 발생 시 rollback
...         conn.execute("UPDATE account SET balance = balance - ? WHERE id = ?", (amount, sender_id))
...         conn.execute("UPDATE account SET balance = balance + ? WHERE id = ?", (amount, receiver_id))
...
>>> transfer(conn, 1, 2, 200)
>>> conn.execute("SELECT id, balance FROM account ORDER BY id").fetchall()
[(1, 800), (2, 700)]

해설:

  • with conn: 블록은 sqlite3에서 가장 간단한 트랜잭션 패턴입니다.
  • 블록 내부가 모두 성공하면 자동 커밋, 예외가 나면 자동 롤백됩니다.
  • 잔액 이동처럼 둘 다 성공하거나 둘 다 실패해야 하는 작업에 필수입니다.

예제 2) 실패를 일부러 만들고 rollback 확인하기

>>> conn = sqlite3.connect(":memory:")
>>> cur = conn.cursor()
>>> cur.execute("CREATE TABLE stock (sku TEXT PRIMARY KEY, qty INTEGER CHECK(qty >= 0))")
<sqlite3.Cursor object at ...>
>>> cur.executemany("INSERT INTO stock(sku, qty) VALUES (?, ?)", [("A", 10), ("B", 5)])
<sqlite3.Cursor object at ...>
>>> conn.commit()

>>> try:
...     with conn:
...         conn.execute("UPDATE stock SET qty = qty - 3 WHERE sku = 'A'")
...         conn.execute("UPDATE stock SET qty = qty - 999 WHERE sku = 'B'")  # CHECK 위반
... except sqlite3.IntegrityError as e:
...     print("rollback:", e)
...
rollback: CHECK constraint failed: qty >= 0

>>> conn.execute("SELECT sku, qty FROM stock ORDER BY sku").fetchall()
[("A", 10), ("B", 5)]

해설:

  • 첫 UPDATE는 성공했지만 두 번째에서 실패했기 때문에 전체 작업이 취소됩니다.
  • 이게 트랜잭션의 핵심 가치입니다. 부분 성공을 허용하지 않음.
  • 제약조건(CHECK, UNIQUE, FOREIGN KEY)과 트랜잭션을 함께 쓰면 데이터 품질이 급격히 좋아집니다.

예제 3) 인덱스 적용 전/후 실행 계획 비교

>>> conn = sqlite3.connect(":memory:")
>>> conn.execute("CREATE TABLE event_log (id INTEGER PRIMARY KEY, user_id INTEGER, event_type TEXT, created_at TEXT)")
<sqlite3.Cursor object at ...>
>>> rows = [(i % 1000, "click" if i % 2 else "view", f"2026-02-17 14:{i%60:02d}:00") for i in range(20000)]
>>> conn.executemany("INSERT INTO event_log(user_id, event_type, created_at) VALUES (?, ?, ?)", rows)
<sqlite3.Cursor object at ...>
>>> conn.commit()

>>> conn.execute("EXPLAIN QUERY PLAN SELECT * FROM event_log WHERE user_id = 777").fetchall()
[(2, 0, 0, 'SCAN event_log')]

>>> conn.execute("CREATE INDEX idx_event_log_user_id ON event_log(user_id)")
<sqlite3.Cursor object at ...>
>>> conn.execute("EXPLAIN QUERY PLAN SELECT * FROM event_log WHERE user_id = 777").fetchall()
[(3, 0, 0, 'SEARCH event_log USING INDEX idx_event_log_user_id (user_id=?)')]

해설:

  • 인덱스 전: SCAN event_log → 테이블 전체 훑기.
  • 인덱스 후: SEARCH ... USING INDEX → 필요한 범위만 탐색.
  • 속도는 데이터 분포/크기마다 다르지만, 실행 계획에서 이미 방향성이 명확히 보입니다.

예제 4) 실무형 패턴: 트랜잭션 + 배치 쓰기 + 조건 인덱스

>>> conn = sqlite3.connect(":memory:")
>>> conn.execute("CREATE TABLE orders (id INTEGER PRIMARY KEY, status TEXT, amount INTEGER, created_at TEXT)")
<sqlite3.Cursor object at ...>
>>> conn.execute("CREATE INDEX idx_orders_status_created ON orders(status, created_at)")
<sqlite3.Cursor object at ...>

>>> payload = [
...     ("PAID", 12000, "2026-02-17 13:00:00"),
...     ("PAID", 9000,  "2026-02-17 13:10:00"),
...     ("CANCEL", 0,   "2026-02-17 13:15:00"),
... ]
>>> with conn:
...     conn.executemany("INSERT INTO orders(status, amount, created_at) VALUES (?, ?, ?)", payload)
...
>>> conn.execute("SELECT COUNT(*) FROM orders WHERE status = 'PAID'").fetchone()[0]
2

해설:

  • 대량 INSERT는 executemany + 단일 트랜잭션이 기본입니다.
  • (status, created_at) 복합 인덱스는 "상태별 최근 주문 조회" 같은 패턴에 효과적입니다.
  • 인덱스 컬럼 순서는 실제 조회 조건 순서와 맞추는 것이 중요합니다.

자주 하는 실수

실수 1) 트랜잭션 없이 여러 UPDATE를 순차 실행

>>> conn = sqlite3.connect(":memory:")
>>> conn.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, v INTEGER)")
<sqlite3.Cursor object at ...>
>>> conn.executemany("INSERT INTO t(id, v) VALUES (?, ?)", [(1, 10), (2, 20)])
<sqlite3.Cursor object at ...>
>>> conn.commit()

>>> # 안 좋은 예: 중간에 예외가 나면 앞 UPDATE만 반영될 수 있음
>>> conn.execute("UPDATE t SET v = v + 100 WHERE id = 1")
<sqlite3.Cursor object at ...>
>>> raise RuntimeError("중간 장애")
Traceback (most recent call last):
...
RuntimeError: 중간 장애

원인:

  • 여러 쿼리가 하나의 논리 작업인데 트랜잭션으로 묶지 않아 부분 반영이 발생합니다.

해결:

>>> with conn:
...     conn.execute("UPDATE t SET v = v + 100 WHERE id = 1")
...     conn.execute("UPDATE t SET v = v + 100 WHERE id = 2")
...
  • "같이 성공/같이 실패"가 필요한 작업은 반드시 with conn:로 감싸세요.

실수 2) 인덱스를 너무 많이 만들거나, 사용 패턴과 안 맞게 생성

증상:

  • 조회는 빨라지지 않았는데 INSERT/UPDATE가 눈에 띄게 느려집니다.

원인:

  • WHERE에 거의 쓰지 않는 컬럼까지 인덱스를 남발했거나, 복합 인덱스 순서가 쿼리 패턴과 다릅니다.

해결:

  • 실제 쿼리 로그 기준으로 인덱스를 만드세요.
  • EXPLAIN QUERY PLAN으로 인덱스가 실제로 사용되는지 확인하세요.
  • 쓰기 비중이 높은 테이블은 최소 인덱스 전략을 우선 검토하세요.

실수 3) 예외를 broad catch로 삼키고 commit/rollback 흐름을 놓침

>>> try:
...     with conn:
...         conn.execute("INSERT INTO orders(status, amount, created_at) VALUES (?, ?, ?)", (None, -1, "bad"))
... except Exception:
...     pass

원인:

  • except Exception: pass는 장애 원인을 지워버립니다.
  • 롤백은 되었더라도, 왜 실패했는지 추적할 근거가 사라집니다.

해결:

>>> import logging
>>> logging.basicConfig(level=logging.ERROR)
>>> try:
...     with conn:
...         conn.execute("INSERT INTO orders(status, amount, created_at) VALUES (?, ?, ?)", (None, -1, "bad"))
... except sqlite3.DatabaseError as e:
...     logging.error("db write failed: %s", e)
  • 예상 가능한 DB 예외를 구체적으로 잡고, 입력 맥락과 함께 로그를 남겨야 합니다.

실무 패턴

  • 입력 검증 규칙

    • DB에 넣기 전에 Python 레벨에서 1차 검증(필수값/범위/타입)을 수행합니다.
    • 제약조건(UNIQUE, CHECK, FOREIGN KEY)을 DB에도 중복 배치해 방어를 이중화합니다.

  • 트랜잭션 경계 설계

    • "업무적으로 한 덩어리"인 작업 단위로 트랜잭션 경계를 잡습니다.
    • 너무 큰 트랜잭션은 락 경쟁과 실패 복구 비용을 키우므로 배치 크기를 조절합니다.

  • 인덱스 운영 규칙

    • 자주 쓰는 조회부터 우선: WHERE status=? AND created_at>=?면 (status, created_at) 우선.
    • 카디널리티가 극단적으로 낮은 컬럼(예: Y/N) 단독 인덱스는 효과가 약할 수 있습니다.
    • 배포 전후 EXPLAIN QUERY PLAN과 간단한 벤치마크를 남겨 "왜 이 인덱스를 만들었는지" 기록합니다.

  • 관측 가능성(Observability)

    • "몇 건 시도 / 몇 건 성공 / 몇 건 실패 / 실패 사유"를 구조화 로그로 남깁니다.
    • 재실행 가능한 입력 단위(batch id, 파일명, 구간 시간)를 같이 기록하면 운영 대응이 쉬워집니다.

오늘의 결론

한 줄 요약: sqlite3 안정화의 핵심은 트랜잭션으로 일관성을 지키고, 인덱스로 필요한 조회만 빠르게 만드는 균형 감각입니다.

기억할 것:

  • 트랜잭션은 "에러 처리 옵션"이 아니라 데이터 무결성의 기본 장치입니다.
  • 인덱스는 많이 만드는 게 아니라, 쿼리 패턴에 맞게 정확히 만드는 것입니다.
  • EXPLAIN QUERY PLAN으로 확인하지 않은 성능 개선은 추측에 가깝습니다.

연습문제

  1. 계좌 이체 함수 transfer()를 작성하세요. 조건: 금액이 0 이하이면 예외, 송금 계좌 잔액 부족이면 예외, 성공 시에만 커밋.
  2. orders(status, created_at, amount) 테이블에서 status='PAID' AND created_at>=... 조회가 자주 발생할 때, 어떤 인덱스를 만들지 설계하고 이유를 적어보세요.
  3. 인덱스 생성 전/후 EXPLAIN QUERY PLAN 결과를 비교해, 어떤 문자열 변화(SCAN→SEARCH)가 나타나는지 확인해 보세요.

이전 강의 정답

  1. 두 문자열의 라인 단위 차이를 unified diff로 출력
>>> import difflib
>>> old = ["name: kim\n", "role: user\n", "active: true\n"]
>>> new = ["name: kim\n", "role: admin\n", "active: true\n"]
>>> print("".join(difflib.unified_diff(old, new, fromfile="old", tofile="new")))
--- old
+++ new
@@ -1,3 +1,3 @@
 name: kim
-role: user
+role: admin
 active: true
  1. 오타 후보를 가장 유사한 키워드로 추천
>>> import difflib
>>> vocab = ["transaction", "rollback", "commit", "index", "integrity"]
>>> difflib.get_close_matches("transection", vocab, n=1, cutoff=0.6)
['transaction']
  1. HTML diff로 변경사항 시각화
>>> import difflib
>>> a = ["line1", "line2-old", "line3"]
>>> b = ["line1", "line2-new", "line3"]
>>> html = difflib.HtmlDiff().make_file(a, b, fromdesc="before", todesc="after")
>>> html.startswith("\n<!DOCTYPE html") or "<html" in html.lower()
True

실습 환경/재현 정보

  • 실행 환경: conda env python100 (Python 3.11.14)
  • 가정한 OS: macOS/Linux 공통
  • SQLite 버전 확인: SELECT sqlite_version();
  • 재현 순서 권장:
    1. :memory: DB에서 트랜잭션 롤백 예제 먼저 실행
    2. 샘플 데이터 1~2만 건 삽입 후 인덱스 전/후 EXPLAIN QUERY PLAN 비교
    3. 실패 케이스(제약 위반)에서 로그와 DB 상태가 기대대로 남는지 확인
  • 체크포인트:
    • 실패 시 부분 반영이 없는지
    • 조회 계획이 SCAN에서 SEARCH ... USING INDEX로 바뀌는지
    • 인덱스 추가 후 쓰기 성능 저하가 허용 범위인지(간단 벤치마크)