Files
wiki/sqlite3/README.md
2024-08-23 08:03:12 -04:00

2.3 KiB

Better Python interface for sqlite3

import sqlite3, json

class DB(sqlite3.Connection):
    def __init__(self, db_name="db.sqlite"):
        super().__init__(db_name)
        with self:
            self.execute('''
                CREATE TABLE IF NOT EXISTS kv_store
                (key TEXT PRIMARY KEY, value BLOB)
            ''')

    def __setitem__(self, key, value):
        if not isinstance(value, bytes): value = json.dumps(value)
        with self:
            cur = self.execute('''
                INSERT OR REPLACE INTO kv_store
                (key, value) VALUES (?, ?)
            ''', (key, value))
            return {"modified_count": cur.rowcount}

    def __getitem__(self, key):
        with self:
            result = self.execute('''
                SELECT value FROM kv_store
                WHERE key = ?
            ''', (key,)).fetchone()
        if result:
            if isinstance(result[0], str):
                try: return json.loads(result[0])
                except json.JSONDecodeError: pass
            return result[0]

    def delete(self, key):
        with self:
            cur = self.execute('''
                DELETE FROM kv_store 
                WHERE key = ?
            ''', (key,))
            return {"deleted_count": cur.rowcount}

    def keys(self, pattern='*'):
        pattern = pattern.translate(str.maketrans({
            '\\': '\\\\', '%': '\\%', '_': '\\_', '*': '%', '?': '_'
        }))
        with self:
            result = self.execute('''
                SELECT key FROM kv_store
                WHERE key LIKE ? ESCAPE '\\'
            ''', (pattern,)).fetchall()
        return [row[0] for row in result]

    def __repr__(self): return repr(self.keys())

Usage

db = DB()

# 문자열 저장 및 조회
db['hello'] = 'world'
print(db['hello'])  # 출력: world (str 타입)

# 숫자 저장 및 조회
db['number'] = 42
print(db['number'])  # 출력: 42 (int 타입)

# 바이너리 데이터 저장 및 조회
db['binary'] = b'binary data'
print(db['binary'])  # 출력: b'binary data' (bytes 타입)

# 복잡한 객체 저장
complex_obj = {'name': 'John', 'age': 30, 'city': 'New York'}
db['complex'] = complex_obj
loaded_obj = db['complex']
print(loaded_obj)  # 출력: {'name': 'John', 'age': 30, 'city': 'New York'} (dict 타입)