sqlite3 — SQLite 데이터베이스를 위한 DB-API 2.0 인터페이스

sqlite3 모듈은 SQLite 데이터베이스를 위한 PEP 249로 규정된 DB-API 2.0 규격에 부합하는 SQL 인터페이스를 제공해요. SQLite는 별도의 서버 프로세스가 필요 없는 가벼운 디스크 기반 데이터베이스를 제공하는 C 라이브러리예요. SQLite로 프로토타입을 만든 뒤 PostgreSQL이나 Oracle 같은 더 큰 데이터베이스로 코드를 옮기는 방식도 가능해요. Sophie에서도 동작하고, 선택적(optional) 모듈이에요.

출처: Python 표준 라이브러리

본문

이 문서는 네 가지 주요 섹션으로 이루어져 있어요. **튜토리얼(Tutorial)**은 sqlite3 모듈 사용법을 가르치고, **레퍼런스(Reference)**는 모듈이 정의하는 클래스와 함수를 설명하며, How-to 가이드는 특정 작업을 처리하는 방법을, **설명(Explanation)**은 트랜잭션 제어에 대한 심층 배경을 다뤄요.

SQLite 자체에 대한 자세한 내용은 https://www.sqlite.org 를 참고하세요.

튜토리얼

이 튜토리얼에서는 기본적인 sqlite3 기능으로 Monty Python 영화 데이터베이스를 만들 거예요. 커서와 트랜잭션을 포함한 데이터베이스 개념의 기본 이해를 전제로 해요.

먼저 새 데이터베이스를 만들고 연결(connection)을 열어요. sqlite3.connect()로 현재 작업 디렉터리의 tutorial.db 데이터베이스에 연결을 만들면, 없으면 암묵적으로 생성돼요:

import sqlite3
con = sqlite3.connect("tutorial.db")

반환된 Connection 객체 con은 디스크의 데이터베이스에 대한 연결을 나타내요. SQL 문을 실행하고 결과를 가져오려면 데이터베이스 커서가 필요해요. con.cursor()로 Cursor를 만들어요:

cur = con.cursor()

이제 title, release year, review score 열을 가진 movie 테이블을 만들 수 있어요. SQLite의 유연한 타입 지정 덕분에 타입 지정은 선택 사항이에요. cur.execute(...)로 CREATE TABLE 문을 실행해요:

cur.execute("CREATE TABLE movie(title, year, score)")

SQLite에 내장된 sqlite_master 테이블을 조회해 새 테이블이 생성됐는지 확인할 수 있어요:

>>> res = cur.execute("SELECT name FROM sqlite_master")
>>> res.fetchone()
('movie',)

존재하지 않는 spam 테이블을 조회하면 res.fetchone()은 None을 반환해요:

>>> res = cur.execute("SELECT name FROM sqlite_master WHERE name='spam'")
>>> res.fetchone() is None
True

INSERT 문으로 두 행의 데이터를 추가해요:

cur.execute("""
    INSERT INTO movie VALUES
        ('Monty Python and the Holy Grail', 1975, 8.2),
        ('And Now for Something Completely Different', 1971, 7.5)
""")

INSERT 문은 트랜잭션을 암묵적으로 열고, 변경 사항이 저장되려면 커밋이 필요해요. con.commit()으로 트랜잭션을 커밋해요:

con.commit()

SELECT 쿼리로 데이터가 올바르게 삽입됐는지 확인해요. res.fetchall()은 결과의 모든 행을 반환해요:

>>> res = cur.execute("SELECT score FROM movie")
>>> res.fetchall()
[(8.2,), (7.5,)]

이제 cur.executemany(...)로 세 행을 더 삽입해요. ? 플레이스홀더를 사용해서 data를 쿼리에 바인딩하는 것에 주목하세요. SQL 주입 공격을 피하려면 항상 문자열 포매팅 대신 플레이스홀더를 사용해야 해요:

data = [
    ("Monty Python Live at the Hollywood Bowl", 1982, 7.9),
    ("Monty Python's The Meaning of Life", 1983, 7.5),
    ("Monty Python's Life of Brian", 1979, 8.0),
]
cur.executemany("INSERT INTO movie VALUES(?, ?, ?)", data)
con.commit()  # Remember to commit the transaction after executing INSERT.

이번에는 쿼리 결과를 반복해서 검증해요:

>>> for row in cur.execute("SELECT year, title FROM movie ORDER BY year"):
...     print(row)
(1971, 'And Now for Something Completely Different')
(1975, 'Monty Python and the Holy Grail')
(1979, "Monty Python's Life of Brian")
(1982, 'Monty Python Live at the Hollywood Bowl')
(1983, "Monty Python's The Meaning of Life")

마지막으로 con.close()로 기존 연결을 닫고 새 연결을 열어 데이터베이스가 디스크에 저장됐는지 확인해요:

>>> con.close()
>>> new_con = sqlite3.connect("tutorial.db")
>>> new_cur = new_con.cursor()
>>> res = new_cur.execute("SELECT title, year FROM movie ORDER BY score DESC")
>>> title, year = res.fetchone()
>>> print(f'The highest scoring Monty Python movie is {title!r}, released in {year}')
The highest scoring Monty Python movie is 'Monty Python and the Holy Grail', released in 1975
>>> new_con.close()

모듈 함수

sqlite3.connect(database, timeout=5.0, detect_types=0, isolation_level='DEFERRED', check_same_thread=True, factory=sqlite3.Connection, cached_statements=128, uri=False, *, autocommit=sqlite3.LEGACY_TRANSACTION_CONTROL) SQLite 데이터베이스에 대한 연결을 열어요. 주요 매개변수:

database (path-like) — 열 데이터베이스 파일 경로. ":memory:"를 넘기면 메모리에만 존재하는 데이터베이스를 만들 수 있어요.
timeout (float) — 테이블이 잠겼을 때 OperationalError를 발생시키기 전까지 연결이 기다리는 초. 기본 5초.
detect_types (int) — register_converter()로 등록한 변환기를 써서 SQLite가 네이티브로 지원하지 않는 데이터 타입을 Python 타입으로 변환할지 제어해요. PARSE_DECLTYPES와 PARSE_COLNAMES를 |(비트 OR)로 조합해요. 두 플래그를 모두 설정하면 열 이름이 선언된 타입보다 우선해요. 기본(0)은 타입 감지 비활성.
isolation_level (str | None) — 레거시 트랜잭션 처리 동작 제어. "DEFERRED"(기본), "EXCLUSIVE", "IMMEDIATE", 또는 암묵 트랜잭션을 비활성화하려면 None. Connection.autocommit이 LEGACY_TRANSACTION_CONTROL(기본)일 때만 효과가 있어요.
check_same_thread (bool) — True(기본)면 연결을 만든 스레드가 아닌 다른 스레드가 사용하면 ProgrammingError를 발생시켜요.
factory (Connection) — 커넥션을 만들 때 쓸 사용자 지정 Connection 서브클래스.
cached_statements (int) — 구문 분석 오버헤드를 줄이기 위해 내부적으로 캐시할 문 수. 기본 128.
uri (bool) — True면 database를 파일 경로와 선택적 쿼리 문자열을 가진 URI로 해석해요. 스킴 부분은 "file:"이어야 해요.
autocommit (bool) — PEP 249 트랜잭션 처리 동작 제어. 현재 기본값은 LEGACY_TRANSACTION_CONTROL이고, 향후 릴리스에서 False로 바뀔 예정이에요.

감사 이벤트 sqlite3.connect(database 인자)와 sqlite3.connect/handle(connection_handle 인자)을 발생시켜요. 3.4에서 uri, 3.10에서 sqlite3.connect/handle 감사 이벤트, 3.12에서 autocommit 추가. 3.13부터 timeout, detect_types, isolation_level, check_same_thread, factory, cached_statements, uri의 위치 인자 사용이 폐기(3.15에서 키워드 전용이 됨).

sqlite3.complete_statement(statement) 문자열 statement가 하나 이상의 완전한 SQL 문을 담고 있으면 True를 반환해요. 닫히지 않은 문자열 리터럴이 없는지와 세미콜론으로 끝나는지 정도만 검사하고, 구문 검증이나 파싱은 하지 않아요.

>>> sqlite3.complete_statement("SELECT foo FROM bar;")
True
>>> sqlite3.complete_statement("SELECT foo")
False

sqlite3.enable_callback_tracebacks(flag, /) 콜백 traceback을 활성화/비활성화해요. 기본적으로 사용자 정의 함수, 집계, 변환기, authorizer 콜백 등에서 traceback을 얻지 못해요. 디버깅하려면 flag=True로 호출해요. 이후 콜백의 traceback이 sys.stderr로 출력돼요. False로 다시 비활성화해요.

sqlite3.register_adapter(type, adapter, /) Python 타입 type을 SQLite 타입으로 적응(adapt)시키는 어댑터 콜러블을 등록해요. 어댑터는 type의 Python 객체 하나를 인자로 받아 SQLite가 네이티브로 이해하는 타입의 값을 반환해야 해요.

sqlite3.register_converter(typename, converter, /) typename 타입의 SQLite 객체를 특정 Python 타입으로 변환하는 콘버터 콜러블을 등록해요. 콘버터는 항상 bytes 객체를 받아 원하는 Python 타입을 반환해요. 타입 감지는 connect()의 detect_types 매개변수로 제어해요. typename과 쿼리의 타입 이름은 대소문자 무시로 매칭돼요.

모듈 상수

sqlite3.LEGACY_TRANSACTION_CONTROL — autocommit에 이 상수를 설정해 구식(Python 3.12 이전) 트랜잭션 제어 동작을 선택해요.
sqlite3.PARSE_DECLTYPES — connect()의 detect_types에 넘겨 각 열의 선언된 타입으로 콘버터 함수를 찾아요. 선언 타입의 첫 단어가 콘버터 딕셔너리 키예요:

CREATE TABLE test(
   i integer primary key,  ! will look up a converter named "integer"
   p point,                ! will look up a converter named "point"
   n number(10)            ! will look up a converter named "number"
 )

sqlite3.PARSE_COLNAMES — detect_types에 넘겨 쿼리 열 이름에서 파싱한 타입 이름으로 콘버터를 찾아요. 열 이름은 큰따옴표(")로, 타입 이름은 대괄호([])로 감싸요:

SELECT MAX(p) as "p [point]" FROM test;  ! will look up converter "point"

sqlite3.SQLITE_OK, sqlite3.SQLITE_DENY, sqlite3.SQLITE_IGNORE — Connection.set_authorizer()에 넘기는 authorizer 콜백이 반환하는 플래그로, 각각 접근 허용, SQL 문을 오류로 중단, 열을 NULL로 취급을 나타내요.
sqlite3.apilevel — 지원하는 DB-API 수준. "2.0"으로 고정.
sqlite3.paramstyle — 모듈이 기대하는 매개변수 마커 형식. "qmark"로 고정(named 스타일도 지원).
sqlite3.sqlite_version — 런타임 SQLite 라이브러리의 버전 번호(string).
sqlite3.sqlite_version_info — 런타임 SQLite 버전(tuple of integers).
sqlite3.threadsafety — 지원하는 스레드 안전 수준. 쿼리 결과: single-thread→0, multi-thread→1, serialized→3 (DB-API 2.0 의미에 따라 0/2/1). 3.11에서 동적으로 설정됨.
sqlite3.SQLITE_DBCONFIG_* — Connection.setconfig()/getconfig() 메서드에 쓰이는 상수 세트(SQLITE_DBCONFIG_DEFENSIVE, SQLITE_DBCONFIG_ENABLE_FKEY, SQLITE_DBCONFIG_TRUSTED_SCHEMA 등). 3.12에 추가됨.
version과 version_info 상수는 3.12에서 폐기, 3.14에서 제거됨.

Connection 객체

class sqlite3.Connection 열린 각 SQLite 데이터베이스는 sqlite3.connect()로 만든 Connection 객체로 표현돼요. 주 목적은 Cursor 객체를 만들고 트랜잭션을 제어하는 거예요. 3.13부터 close()를 호출하지 않고 삭제하면 ResourceWarning이 발생.

주요 메서드:

cursor(factory=Cursor) — Cursor 객체를 만들어 반환해요. factory는 Cursor나 그 서브클래스 인스턴스를 반환하는 콜러블이면 돼요.
blobopen(table, column, rowid, /, *, readonly=False, name='main') — 기존 BLOB에 대한 Blob 핸들을 열어요. WITHOUT ROWID 테이블에서 열면 OperationalError. 3.11 추가.
commit() — 보류 중인 트랜잭션을 데이터베이스에 커밋해요. autocommit이 True이거나 열린 트랜잭션이 없으면 아무것도 하지 않아요.
rollback() — 보류 중인 트랜잭션의 시작으로 롤백해요.
close() — 연결을 닫아요. autocommit이 False면 보류 트랜잭션이 암묵적으로 롤백돼요. 닫기 전에 commit()해서 변경을 잃지 않도록 하세요.
execute(sql, parameters=(), /), executemany(sql, parameters, /), executescript(sql_script, /) — 새 Cursor를 만들고 그 위에서 각각 execute()/executemany()/executescript()를 호출하고 새 커서를 반환해요(연결 단축 메서드).
create_function(name, narg, func, *, deterministic=False) — 사용자 정의 SQL 함수를 만들거나 제거해요. func는 SQLite가 네이티브 지원하는 타입을 반환해야 해요. None으로 설정하면 기존 함수를 제거해요. deterministic=True면 SQLite가 추가 최적화를 수행해요.

>>> import hashlib
>>> def md5sum(t):
...     return hashlib.md5(t).hexdigest()
>>> con = sqlite3.connect(":memory:")
>>> con.create_function("md5", 1, md5sum)
>>> for row in con.execute("SELECT md5(?)", (b"foo",)):
...     print(row)
('acbd18db4cc2f85cedef654fccc4a4d8',)
>>> con.close()

create_aggregate(name, n_arg, aggregate_class) — 사용자 정의 SQL 집계 함수를 만들거나 제거해요. aggregate_class는 step()(집계에 행 추가)과 finalize()(SQLite 네이티브 타입으로 최종 결과 반환) 메서드를 구현해야 해요.

class MySum:
    def __init__(self):
        self.count = 0

    def step(self, value):
        self.count += value

    def finalize(self):
        return self.count

con = sqlite3.connect(":memory:")
con.create_aggregate("mysum", 1, MySum)
cur = con.execute("CREATE TABLE test(i)")
cur.execute("INSERT INTO test(i) VALUES(1)")
cur.execute("INSERT INTO test(i) VALUES(2)")
cur.execute("SELECT mysum(i) FROM test")
print(cur.fetchone()[0])

con.close()

create_window_function(name, num_params, aggregate_class, /) — 사용자 정의 집계 윈도우 함수를 만들거나 제거해요. aggregate_class는 step(), value(), inverse(), finalize()를 구현해야 해요. SQLite 3.25.0 미만이면 NotSupportedError. 3.11 추가.
create_collation(name, callable, /) — 콜레이션 함수 callable로 이름이 name인 콜레이션을 만들어요. 콜러블은 두 string 인자를 받아 첫 번째가 더 높으면 1, 낮으면 -1, 같으면 0을 반환해요:

def collate_reverse(string1, string2):
    if string1 == string2:
        return 0
    elif string1 < string2:
        return 1
    else:
        return -1

con = sqlite3.connect(":memory:")
con.create_collation("reverse", collate_reverse)

cur = con.execute("CREATE TABLE test(x)")
cur.executemany("INSERT INTO test(x) VALUES(?)", [("a",), ("b",)])
cur.execute("SELECT x FROM test ORDER BY x COLLATE reverse")
for row in cur:
    print(row)
con.close()

callable을 None으로 설정하면 콜레이션을 제거해요.

interrupt() — 다른 스레드에서 호출해 이 연결에서 실행 중일 수 있는 쿼리를 중단해요. 중단된 쿼리는 OperationalError를 발생시켜요.
set_authorizer(authorizer_callback) — 테이블 열에 접근할 때마다 호출될 콜러블을 등록해요. 콜백은 SQLITE_OK, SQLITE_DENY, SQLITE_IGNORE 중 하나를 반환해야 해요. None 전달로 비활성화. 3.11에서 None 지원.
set_progress_handler(progress_handler, n) — SQLite 가상 머신 명령어마다 호출될 핸들러를 등록해요. 핸들러가 0이 아닌 값을 반환하면 현재 쿼리를 종료하고 DatabaseError를 발생시켜요.
set_trace_callback(trace_callback) — SQLite 백엔드가 실제 실행하는 각 SQL 문에 대해 호출될 콜백을 등록해요. 3.3 추가.
enable_load_extension(enabled, /) — True면 SQLite 엔진이 공유 라이브러리에서 SQLite 확장을 로드하도록 허용해요. sqlite3 모듈은 기본적으로 로드 가능한 확장을 지원하도록 빌드되지 않아요(일부 플랫폼, 특히 macOS는 이 기능 없이 컴파일됨). 감사 이벤트 sqlite3.enable_load_extension(connection, enabled). 3.2 추가.
load_extension(path, /, *, entrypoint=None) — 공유 라이브러리에서 SQLite 확장을 로드해요. 3.12에서 entrypoint 추가.
iterdump(*, filter=None) — 데이터베이스를 SQL 소스 코드로 덤프하는 이터레이터를 반환해요. filter는 LIKE 패턴(예: prefix_%)이 될 수 있어요. 3.13에서 filter 추가.

# Convert file example.db to SQL dump file dump.sql
con = sqlite3.connect('example.db')
with open('dump.sql', 'w') as f:
    for line in con.iterdump():
        f.write('%s\n' % line)
con.close()

backup(target, *, pages=-1, progress=None, name='main', sleep=0.250) — SQLite 데이터베이스의 백업을 만들어요. 다른 클라이언트가 접근 중이거나 같은 연결이 동시에 쓰고 있어도 동작해요.

def progress(status, remaining, total):
    print(f'Copied {total-remaining} of {total} pages...')

src = sqlite3.connect('example.db')
dst = sqlite3.connect('backup.db')
with dst:
    src.backup(dst, pages=1, progress=progress)
dst.close()
src.close()

3.7 추가.

getlimit(category, /), setlimit(category, limit, /) — 연결 런타임 한계를 조회/설정해요. SQLITE_LIMIT_* 상수를 사용해요.

>>> con.getlimit(sqlite3.SQLITE_LIMIT_SQL_LENGTH)
1000000000

3.11 추가.

getconfig(op, /), setconfig(op, enable=True, /) — 불리언 연결 구성 옵션을 조회/설정해요. SQLITE_DBCONFIG_* 코드 사용. 3.12 추가.
serialize(*, name='main') — 데이터베이스를 bytes 객체로 직렬화해요. 3.11 추가.
deserialize(data, /, *, name='main') — 직렬화된 데이터베이스를 Connection으로 역직렬화해요. 3.11 추가.

주요 속성:

autocommit — PEP 249 호환 트랜잭션 동작 제어. 허용 값: False(권장, 트랜잭션이 항상 열려 있음), True(SQLite autocommit 모드), LEGACY_TRANSACTION_CONTROL(현재 기본). 3.12 추가.
in_transaction — 트랜잭션이 활성이면(True) True. 3.2 추가.
isolation_level — 레거시 트랜잭션 처리 모드 제어. None이면 트랜잭션을 암묵적으로 열지 않아요. autocommit이 LEGACY_TRANSACTION_CONTROL일 때만 효과가 있어요.
row_factory — 이 연결에서 만든 Cursor 객체의 초기 row_factory. 기본 None(각 행이 tuple). 3.14.6에서 삭제 불가.
text_factory — TEXT 데이터 타입의 SQLite 값에 대해 bytes를 받아 텍스트 표현을 반환하는 콜러블. 기본 str. 3.14.6에서 삭제 불가.
total_changes — 연결이 열린 이후 수정/삽입/삭제된 총 데이터베이스 행 수.

Cursor 객체

class sqlite3.Cursor Cursor 객체는 SQL 문을 실행하고 fetch 작업의 컨텍스트를 관리하는 데 쓰이는 데이터베이스 커서를 나타내요. Connection.cursor()나 연결 단축 메서드로 만들어져요. 커서 객체는 이터레이터라서 SELECT 쿼리를 실행하면 커서를 그대로 반복해 결과 행을 가져올 수 있어요:

for row in cur.execute("SELECT t FROM data"):
    print(row)

주요 메서드:

execute(sql, parameters=(), /) — 단일 SQL 문을 실행해요. parameters는 named 플레이스홀더면 dict, unnamed면 시퀀스예요. sql이 둘 이상의 문이거나 named 플레이스홀더인데 parameters가 시퀀스면 ProgrammingError. 3.14 변경.
executemany(sql, parameters, /) — parameters의 각 항목에 대해 매개변수화된 DML 문 sql을 반복 실행해요.
executescript(sql_script, /) — sql_script의 SQL 문들을 실행해요. sql_script는 string이어야 해요.
fetchone() — 쿼리 결과 집합의 다음 행을 tuple로 반환하고, 더 이상 데이터가 없으면 None을 반환해요.
fetchmany(size=cursor.arraysize) — 다음 행 세트를 list로 반환해요. 더 없으면 빈 리스트. 3.14.1에서 음수 size는 ValueError.
fetchall() — (남은) 모든 행을 list로 반환해요.
close() — 커서를 닫아요. 이후 사용 시 ProgrammingError.
setinputsizes(sizes, /), setoutputsize(size, column=None, /) — DB-API 요구 사항. sqlite3에서는 아무것도 하지 않아요.

주요 속성:

arraysize — fetchmany()가 반환할 행 수 제어. 기본 1.
connection — 커서가 속한 Connection. 읽기 전용.
description — 마지막 쿼리의 열 이름. DB-API 호환을 위해 각 열마다 마지막 여섯 항목이 None인 7-튜플을 반환해요.
lastrowid — 마지막으로 삽입된 행의 row id. execute()로 INSERT/REPLACE가 성공한 후에만 갱신돼요. 초기값 None.
rowcount — INSERT/UPDATE/DELETE/REPLACE 문의 수정 행 수. 다른 문은 -1. execute()/executemany()로만 갱신되고, 결과 행을 모두 가져와야 갱신돼요.
row_factory — 이 커서에서 가져온 행의 표현 제어. None이면 tuple, sqlite3.Row나 사용자 지정 콜러블로 설정할 수 있어요. 3.14.6에서 삭제 불가.

Row 객체

class sqlite3.Row Connection 객체를 위한 고도로 최적화된 row_factory로 쓰이는 인스턴스예요. 반복, 동등성 검사, len(), 열 이름과 인덱스로의 매핑 접근을 지원해요. 두 Row는 열 이름과 값이 같으면 같다고 비교돼요.

keys() — 열 이름을 string 리스트로 반환해요. 3.5에서 슬라이싱 지원.

Blob 객체

class sqlite3.Blob 3.11 추가. SQLite BLOB에서 데이터를 읽고 쓸 수 있는 파일류 객체예요. len(blob)으로 크기를 얻고, 인덱스/슬라이스로 직접 접근해요. 컨텍스트 매니저로 써서 사용 후 핸들이 닫히도록 보장할 수 있어요:

con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE test(blob_col blob)")
con.execute("INSERT INTO test(blob_col) VALUES(zeroblob(13))")

# Write to our blob, using two write operations:
with con.blobopen("test", "blob_col", 1) as blob:
    blob.write(b"hello, ")
    blob.write(b"world.")
    # Modify the first and last bytes of our blob
    blob[0] = ord("H")
    blob[-1] = ord("!")

# Read the contents of our blob
with con.blobopen("test", "blob_col", 1) as blob:
    greeting = blob.read()

print(greeting)  # outputs "b'Hello, world!'"
con.close()

close(), read(length=-1, /), write(data, /)(끝을 넘어 쓰면 ValueError), tell(), seek(offset, origin=os.SEEK_SET, /).

PrepareProtocol 객체

class sqlite3.PrepareProtocol 자신을 네이티브 SQLite 타입으로 적응시킬 수 있는 객체를 위한 PEP 246 스타일 적응 프로토콜로 작동하는 것이 유일한 목적이에요.

예외

예외 계층은 DB-API 2.0(PEP 249)이 정의해요:

sqlite3.Warning — Exception의 서브클래스. 현재 모듈에서는 발생하지 않지만 응용이 발생시킬 수 있어요.
sqlite3.Error — 모듈의 다른 예외들의 기본 클래스. 모든 오류를 except 하나로 잡으려면 이걸 사용해요.
sqlite3.InterfaceError — 저수준 SQLite C API 남용 시. Error의 서브클래스.
sqlite3.DatabaseError — 데이터베이스 관련 오류. Error의 서브클래스. 특수화된 서브클래스를 통해서만 암묵적으로 발생해요.
sqlite3.DataError — 처리된 데이터 문제(숫자 범위 초과, 문자열이 너무 긺 등). DatabaseError의 서브클래스.
sqlite3.OperationalError — 데이터베이스 운영 관련 오류(경로를 찾을 수 없음, 트랜잭션 처리 실패 등). DatabaseError의 서브클래스.
sqlite3.IntegrityError — 관계형 무결성 침해(외래 키 검사 실패 등). DatabaseError의 서브클래스.
sqlite3.InternalError — SQLite의 내부 오류. DatabaseError의 서브클래스.
sqlite3.ProgrammingError — sqlite3 API 프로그래밍 오류(잘못된 바인딩 수, 닫힌 연결 연산 등). DatabaseError의 서브클래스.
sqlite3.NotSupportedError — 기본 SQLite 라이브러리가 지원하지 않는 메서드/API 사용 시. DatabaseError의 서브클래스.

SQLite 라이브러리 내부에서 오류가 발생하면 예외에 sqlite_errorcode(숫자 코드)와 sqlite_errorname(기호 이름) 속성이 추가돼요. 3.11 추가.

SQLite와 Python 타입

SQLite는 NULL, INTEGER, REAL, TEXT, BLOB을 네이티브로 지원해요. 문제없이 보낼 수 있는 Python 타입: None→NULL, int→INTEGER, float→REAL, str→TEXT, bytes→BLOB.

기본 변환: NULL→None, INTEGER→int, REAL→float, TEXT→text_factory에 따라 다름(기본 str), BLOB→bytes.

sqlite3 모듈의 타입 시스템은 두 가지로 확장 가능해요: 객체 어댑터로 추가 Python 타입을 저장하거나, 콘버터로 SQLite 타입을 Python 타입으로 변환시킬 수 있어요.

참고: 기본 어댑터와 콘버터는 Python 3.12부터 폐기됐어요. 대신 Adapter/converter 레시피를 사용하세요. 폐기된 것들: datetime.date/datetime.datetime을 ISO 8601 문자열로, 선언된 "date" 타입을 datetime.date 객체로, "timestamp" 타입을 datetime.datetime 객체로 변환하는 것들.

명령줄 인터페이스

sqlite3 모듈은 인터프리터의 -m 스위치로 스크립트처럼 호출해 간단한 SQLite 셸을 제공할 수 있어요:

python -m sqlite3 [-h] [-v] [filename] [sql]

.quit 또는 CTRL-D로 셸을 종료해요. -h, --help는 CLI 도움말, -v, --version은 기본 SQLite 라이브러리 버전을 출력해요. 3.12 추가.

How-to: 플레이스홀더로 값 바인딩

SQL 작업은 보통 Python 변수의 값을 써야 해요. 하지만 Python의 문자열 연산으로 쿼리를 조립하면 SQL 주입 공격에 취약해요. 예를 들어 공격자는 작은따옴표를 닫고 OR TRUE를 주입해 모든 행을 선택할 수 있어요:

>>> # Never do this -- insecure!
>>> symbol = input()
' OR TRUE; --
>>> sql = "SELECT * FROM stocks WHERE symbol = '%s'" % symbol
>>> print(sql)
SELECT * FROM stocks WHERE symbol = '' OR TRUE; --'
>>> cur.execute(sql)

대신 DB-API의 매개변수 치환을 사용해요. qmark(?) 또는 named(:name) 스타일의 플레이스홀더를 쓰고, 실제 값을 커서의 execute() 두 번째 인자로 제공해요:

con = sqlite3.connect(":memory:")
cur = con.execute("CREATE TABLE lang(name, first_appeared)")

# This is the named style used with executemany():
data = (
    {"name": "C", "year": 1972},
    {"name": "Fortran", "year": 1957},
    {"name": "Python", "year": 1991},
    {"name": "Go", "year": 2009},
)
cur.executemany("INSERT INTO lang VALUES(:name, :year)", data)

# This is the qmark style used in a SELECT query:
params = (1972,)
cur.execute("SELECT * FROM lang WHERE first_appeared = ?", params)
print(cur.fetchall())
con.close()

PEP 249 숫자 플레이스홀더는 지원되지 않아요. 쓰면 named 플레이스홀더로 해석돼요.

How-to: 커스텀 Python 타입 적응

객체가 스스로 적응하게 하거나, 어댑터 콜러블을 쓰는 두 가지 방법이 있어요(후자가 우선). __conform__ 메서드로 객체를 적응시킬 수 있어요:

class Point:
    def __init__(self, x, y):
        self.x, self.y = x, y

    def __conform__(self, protocol):
        if protocol is sqlite3.PrepareProtocol:
            return f"{self.x};{self.y}"

con = sqlite3.connect(":memory:")
cur = con.cursor()

cur.execute("SELECT ?", (Point(4.0, -3.2),))
print(cur.fetchone()[0])
con.close()

또는 register_adapter()로 어댑터 함수를 등록해요:

class Point:
    def __init__(self, x, y):
        self.x, self.y = x, y

def adapt_point(point):
    return f"{point.x};{point.y}"

sqlite3.register_adapter(Point, adapt_point)

con = sqlite3.connect(":memory:")
cur = con.cursor()

cur.execute("SELECT ?", (Point(1.0, 2.5),))
print(cur.fetchone()[0])
con.close()

How-to: SQLite 값을 커스텀 Python 타입으로 변환

콘버터 함수로 SQLite 값을 Python 타입으로 변환할 수 있어요. 콘버터 함수는 항상 bytes 객체를 받아요:

def convert_point(s):
    x, y = map(float, s.split(b";"))
    return Point(x, y)

언제 변환할지는 연결 시 detect_types 매개변수로 정해요. PARSE_DECLTYPES(암묵), PARSE_COLNAMES(명시), PARSE_DECLTYPES | PARSE_COLNAMES(둘 다) 중 선택하세요. 어댑터/콘버터 레시피:

import datetime as dt
import sqlite3

def adapt_date_iso(val):
    """Adapt datetime.date to ISO 8601 date."""
    return val.isoformat()

def adapt_datetime_iso(val):
    """Adapt datetime.datetime to timezone-naive ISO 8601 date."""
    return val.replace(tzinfo=None).isoformat()

def adapt_datetime_epoch(val):
    """Adapt datetime.datetime to Unix timestamp."""
    return int(val.timestamp())

sqlite3.register_adapter(dt.date, adapt_date_iso)
sqlite3.register_adapter(dt.datetime, adapt_datetime_iso)
sqlite3.register_adapter(dt.datetime, adapt_datetime_epoch)

def convert_date(val):
    """Convert ISO 8601 date to datetime.date object."""
    return dt.date.fromisoformat(val.decode())

def convert_datetime(val):
    """Convert ISO 8601 datetime to datetime.datetime object."""
    return dt.datetime.fromisoformat(val.decode())

def convert_timestamp(val):
    """Convert Unix epoch timestamp to datetime.datetime object."""
    return dt.datetime.fromtimestamp(int(val))

sqlite3.register_converter("date", convert_date)
sqlite3.register_converter("datetime", convert_datetime)
sqlite3.register_converter("timestamp", convert_timestamp)

How-to: 연결 단축 메서드

Connection의 execute()/executemany()/executescript() 메서드를 쓰면 종종 불필요한 Cursor 객체를 명시적으로 만들 필요가 없어져요. 커서가 암묵적으로 생성되고 단축 메서드가 그 커서를 반환해요:

# Create and fill the table.
con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE lang(name, first_appeared)")
data = [
    ("C++", 1985),
    ("Objective-C", 1984),
]
con.executemany("INSERT INTO lang(name, first_appeared) VALUES(?, ?)", data)

# Print the table contents
for row in con.execute("SELECT name, first_appeared FROM lang"):
    print(row)

print("I just deleted", con.execute("DELETE FROM lang").rowcount, "rows")

# close() is not a shortcut method and it's not called automatically;
# the connection object should be closed manually
con.close()

How-to: 연결 컨텍스트 매니저

Connection 객체는 컨텍스트 매니저로 쓸 수 있어요. with 본문이 예외 없이 끝나면 트랜잭션을 커밋하고, 커밋 실패나 잡히지 않은 예외가 있으면 롤백해요:

con = sqlite3.connect(":memory:")
con.execute("CREATE TABLE lang(id INTEGER PRIMARY KEY, name VARCHAR UNIQUE)")

# Successful, con.commit() is called automatically afterwards
with con:
    con.execute("INSERT INTO lang(name) VALUES(?)", ("Python",))

# con.rollback() is called after the with block finishes with an exception,
# the exception is still raised and must be caught
try:
    with con:
        con.execute("INSERT INTO lang(name) VALUES(?)", ("Python",))
except sqlite3.IntegrityError:
    print("couldn't add Python twice")

# Connection object used as context manager only commits or rollbacks transactions,
# so the connection object should be closed manually
con.close()

컨텍스트 매니저는 새 트랜잭션을 암묵적으로 열지도, 연결을 닫지도 않아요. 닫는 컨텍스트 매니저가 필요하면 contextlib.closing()을 고려하세요.

How-to: SQLite URIs

유용한 URI 기법들:

읽기 전용으로 열기: sqlite3.connect("file:tutorial.db?mode=ro", uri=True)
새 파일을 암묵적으로 만들지 않기: "file:nosuchdb.db?mode=rw" — 만들 수 없으면 OperationalError
공유 named in-memory 데이터베이스:

db = "file:mem1?mode=memory&cache=shared"
con1 = sqlite3.connect(db, uri=True)
con2 = sqlite3.connect(db, uri=True)
with con1:
    con1.execute("CREATE TABLE shared(data)")
    con1.execute("INSERT INTO shared VALUES(28)")
res = con2.execute("SELECT data FROM shared")
assert res.fetchone() == (28,)

con1.close()
con2.close()

How-to: row factory

기본적으로 sqlite3는 각 행을 tuple로 나타내요. sqlite3.Row 클래스나 커스텀 row_factory를 쓸 수 있어요. Connection.row_factory를 설정하는 게 권장돼요. Row는 tuple에 비해 메모리 오버헤드와 성능 영향이 거의 없이 열 인덱스 및 대소문자 무시 이름 접근을 제공해요:

>>> con = sqlite3.connect(":memory:")
>>> con.row_factory = sqlite3.Row

>>> res = con.execute("SELECT 'Earth' AS name, 6378 AS radius")
>>> row = res.fetchone()
>>> row.keys()
['name', 'radius']
>>> row[0]         # Access by index.
'Earth'
>>> row["name"]    # Access by name.
'Earth'
>>> row["RADIUS"]  # Column names are case-insensitive.
6378
>>> con.close()

각 행을 dict로 반환하는 커스텀 팩토리:

def dict_factory(cursor, row):
    fields = [column[0] for column in cursor.description]
    return {key: value for key, value in zip(fields, row)}

named tuple 팩토리:

from collections import namedtuple

def namedtuple_factory(cursor, row):
    fields = [column[0] for column in cursor.description]
    cls = namedtuple("Row", fields)
    return cls._make(row)

How-to: 비-UTF-8 텍스트 인코딩 처리

기본적으로 sqlite3는 TEXT 데이터 타입 값을 적응시키는 데 str을 써요. 다른 인코딩이나 잘못된 UTF-8에서는 실패할 수 있어요. 커스텀 text_factory를 쓰세요:

con.text_factory = lambda data: str(data, encoding="latin2")

잘못된 UTF-8이나 임의 데이터에는 surrogateescape를 써요:

con.text_factory = lambda data: str(data, errors="surrogateescape")

sqlite3 모듈 API는 서로게이트를 포함하는 문자열을 지원하지 않아요.

설명: 트랜잭션 제어

sqlite3는 트랜잭션을 언제, 어떻게 열고 닫을지 제어하는 여러 방법을 제공해요. autocommit 속성을 통한 트랜잭션 제어가 권장되고, isolation_level 속성은 Python 3.12 이전 동작을 유지해요.

autocommit 속성으로 제어(권장): False로 설정하면 PEP 249 호환 트랜잭션 제어를 의미해요. sqlite3가 트랜잭션을 항상 열어 두도록 보장하고, commit()/rollback()으로 명시적으로 닫아요. BEGIN DEFERRED 문을 사용해요. True로 설정하면 SQLite의 autocommit 모드를 켜고, Connection.commit()/rollback()은 효과가 없어요. LEGACY_TRANSACTION_CONTROL로 설정하면 isolation_level에 맡겨요.

isolation_level 속성으로 제어(레거시): Connection.autocommit이 LEGACY_TRANSACTION_CONTROL(기본)일 때 유효해요. isolation_level이 None이 아니면 execute()와 executemany()가 INSERT/UPDATE/DELETE/REPLACE 문을 실행하기 전에 새 트랜잭션이 암묵적으로 열려요. None이면 트랜잭션이 전혀 암묵적으로 열리지 않아요. executescript()는 isolation_level 값과 무관하게 실행 전에 보류 트랜잭션을 암묵적으로 커밋해요. 3.6에서 변경: DDL 문 전에 트랜잭션을 암묵적으로 커밋하던 동작 제거.

더 알아보기

SQLite 공식 문서: https://www.sqlite.org — 지원되는 SQL 방언의 구문과 데이터 타입.
PEP 249 — Database API Specification 2.0.
W3Schools SQL 튜토리얼: https://www.w3schools.com/sql/