Python sqlite3 Patterns

import sqlite3
 
# Auto-commit/rollback with context manager
with sqlite3.connect('app.db') as conn:
    conn.execute('INSERT INTO users (name) VALUES (?)', ('Alice',))
    # Auto-commits on success
    # Auto-rolls back on exception

import sqlite3
 
conn = sqlite3.connect('app.db')
 
# Dictionary rows
conn.row_factory = sqlite3.Row
 
cursor = conn.execute('SELECT * FROM users')
row = cursor.fetchone()
print(row['name'])      # Access by column name
print(dict(row))        # Convert to dict
print(row.keys())       # Column names
 
# Named tuple rows
def namedtuple_factory(cursor, row):
    from collections import namedtuple
    fields = [col[0] for col in cursor.description]
    Row = namedtuple('Row', fields)
    return Row(*row)
 
conn.row_factory = namedtuple_factory

import sqlite3
 
conn = sqlite3.connect('app.db')
 
# Positional (?)
conn.execute('INSERT INTO users (name, age) VALUES (?, ?)', ('Alice', 30))
 
# Named (:name)
conn.execute(
    'INSERT INTO users (name, age) VALUES (:name, :age)',
    {'name': 'Bob', 'age': 25}
)
 
# executemany for batch inserts
users = [('Alice', 30), ('Bob', 25), ('Charlie', 35)]
conn.executemany('INSERT INTO users (name, age) VALUES (?, ?)', users)

import sqlite3
 
conn = sqlite3.connect('app.db')
 
# Explicit transaction
conn.execute('BEGIN')
try:
    conn.execute('UPDATE accounts SET balance = balance - 100 WHERE id = 1')
    conn.execute('UPDATE accounts SET balance = balance + 100 WHERE id = 2')
    conn.execute('COMMIT')
except:
    conn.execute('ROLLBACK')
    raise
 
# With isolation level
conn = sqlite3.connect('app.db', isolation_level='DEFERRED')
# Options: None (autocommit), 'DEFERRED', 'IMMEDIATE', 'EXCLUSIVE'

import sqlite3
from datetime import datetime
import json
 
# Register adapter (Python → SQLite)
def adapt_datetime(dt):
    return dt.isoformat()
 
sqlite3.register_adapter(datetime, adapt_datetime)
 
# Register converter (SQLite → Python)
def convert_datetime(data):
    return datetime.fromisoformat(data.decode())
 
sqlite3.register_converter('DATETIME', convert_datetime)
 
# Enable type detection
conn = sqlite3.connect('app.db', detect_types=sqlite3.PARSE_DECLTYPES)
 
# JSON column
sqlite3.register_adapter(dict, json.dumps)
sqlite3.register_converter('JSON', lambda x: json.loads(x.decode()))

import sqlite3
 
# Ephemeral in-memory database
conn = sqlite3.connect(':memory:')
 
# Shared in-memory (multiple connections)
conn = sqlite3.connect('file::memory:?cache=shared', uri=True)

import sqlite3
 
conn = sqlite3.connect('app.db')
 
# Create FTS5 table
conn.execute('''
    CREATE VIRTUAL TABLE IF NOT EXISTS articles_fts 
    USING fts5(title, content)
''')
 
# Insert
conn.execute('INSERT INTO articles_fts VALUES (?, ?)', 
             ('Python Guide', 'Learn Python programming...'))
 
# Search
results = conn.execute('''
    SELECT * FROM articles_fts WHERE articles_fts MATCH 'python'
''').fetchall()
 
# Ranked search
results = conn.execute('''
    SELECT *, rank FROM articles_fts 
    WHERE articles_fts MATCH 'python' 
    ORDER BY rank
''').fetchall()

import sqlite3
 
conn = sqlite3.connect('app.db')
 
# Store and query JSON
conn.execute('''
    CREATE TABLE IF NOT EXISTS events (
        id INTEGER PRIMARY KEY,
        data JSON
    )
''')
 
conn.execute('INSERT INTO events (data) VALUES (?)', 
             ['{"type": "click", "x": 100, "y": 200}'])
 
# Extract JSON values
result = conn.execute('''
    SELECT json_extract(data, '$.type') as event_type 
    FROM events
''').fetchone()

import sqlite3
from queue import Queue
from contextlib import contextmanager
 
class SQLitePool:
    def __init__(self, database, size=5):
        self.database = database
        self.pool = Queue(maxsize=size)
        for _ in range(size):
            conn = sqlite3.connect(database, check_same_thread=False)
            conn.row_factory = sqlite3.Row
            self.pool.put(conn)
    
    @contextmanager
    def connection(self):
        conn = self.pool.get()
        try:
            yield conn
        finally:
            self.pool.put(conn)
 
pool = SQLitePool('app.db')
with pool.connection() as conn:
    conn.execute('SELECT * FROM users')

import sqlite3
 
source = sqlite3.connect('app.db')
backup = sqlite3.connect('backup.db')
 
source.backup(backup)
 
# Or with progress callback
def progress(status, remaining, total):
    print(f'Copied {total-remaining}/{total} pages')
 
source.backup(backup, pages=10, progress=progress)

import sqlite3
 
conn = sqlite3.connect('app.db')
conn.execute('PRAGMA journal_mode=WAL')
 
# Better concurrent read performance
# Reads don't block writes and vice versa

import sqlite3
 
conn = sqlite3.connect('app.db')
 
# Use executemany for bulk inserts
data = [(i, f'user{i}') for i in range(10000)]
conn.executemany('INSERT INTO users VALUES (?, ?)', data)
 
# Indexes
conn.execute('CREATE INDEX IF NOT EXISTS idx_users_name ON users(name)')
 
# Analyze for query planner
conn.execute('ANALYZE')
 
# Vacuum to reclaim space
conn.execute('VACUUM')
 
# Cache size (pages)
conn.execute('PRAGMA cache_size = 10000')
 
# Synchronous mode (careful with data safety)
conn.execute('PRAGMA synchronous = NORMAL')

import sqlite3
 
conn = sqlite3.connect('app.db')
 
conn.execute('''
    INSERT INTO users (id, name, visits) VALUES (?, ?, 1)
    ON CONFLICT(id) DO UPDATE SET visits = visits + 1
''', (1, 'Alice'))

import sqlite3
import threading
 
# Default: connections can't be shared between threads
conn = sqlite3.connect('app.db')  # Only use in one thread
 
# Allow multi-thread access (not recommended)
conn = sqlite3.connect('app.db', check_same_thread=False)
 
# Better: one connection per thread
local = threading.local()
 
def get_conn():
    if not hasattr(local, 'conn'):
        local.conn = sqlite3.connect('app.db')
    return local.conn

import sqlite3
from dataclasses import dataclass
from typing import Optional, List
 
@dataclass
class User:
    id: Optional[int]
    name: str
    email: str
 
class UserRepository:
    def __init__(self, conn: sqlite3.Connection):
        self.conn = conn
        self.conn.row_factory = sqlite3.Row
    
    def get(self, user_id: int) -> Optional[User]:
        row = self.conn.execute(
            'SELECT * FROM users WHERE id = ?', (user_id,)
        ).fetchone()
        return User(**dict(row)) if row else None
    
    def save(self, user: User) -> User:
        if user.id is None:
            cursor = self.conn.execute(
                'INSERT INTO users (name, email) VALUES (?, ?)',
                (user.name, user.email)
            )
            user.id = cursor.lastrowid
        else:
            self.conn.execute(
                'UPDATE users SET name=?, email=? WHERE id=?',
                (user.name, user.email, user.id)
            )
        return user