Python SQLite3 usage tutorial

SQLite is the most widely used embedded database engine in the world. It does not have the client/server architecture of a traditional database. Instead, it compresses the entire database system—including table structures, data, and indexes—into a cross-platform common file, which is very suitable for lightweight projects.

This tutorial will take you to master Python built-in modules from scratchsqlite3Core usage: Get started quickly with zero configuration, to advanced-features such as transactions and custom row factories, and finally consolidate what you have learned with complete examples and exercises.

1. Introduction to SQLite

SQLite has several major advantages that make it the lightweight database of choice for developers:

  • 🚫 Zero server, zero configuration: No need to install database software, create one.dbThe file can be used.
  • 🌐 Cross-platform: From mobile phones to servers, all mainstream platforms are available.
  • 📝 Compatible with standard SQL: The learning cost is extremely low, and SQL syntax can be reused.
  • 🐍 Python native built-in: Python 2.5 and above versions come directly with itsqlite3Module, no additional installation required.

2. Pre-requisite core concepts

Before formally writing code, first understand the two core objects and a common operation process.

2.1 Core Objects

Object nameFunction
ConnectionRepresents a physical connection to a database file and is responsible for committing/rolling back transactions and closing the connection.
CursorAttached to Connection, used to execute SQL statements and obtain query results, and can be understood as the "console" of SQL.

2.2 General operating procedures

No matter what operation you do, roughly follow these 6 steps (it will be simplified later using the context manager):

  1. Establish a connection to the database file
  2. Get the cursor object from the connection
  3. Execute SQL statements through cursors
  4. If it is a write operation (add, delete, modify), the transaction needs to be submitted
  5. Close the cursor (recommended)
  6. Close the connection (must be closed to avoid resource leakage)

3. Basic CRUD operations

We use the example of "User Management" to demonstrate addition, deletion, modification and query step by step.

3.1 Connect to database

import sqlite3

# 连接到 example.db 文件（不存在则自动在当前目录创建）
conn = sqlite3.connect('example.db')

If you just want to test temporarily and don't want to generate files, you can use:memory:Create an memory database and the data will disappear automatically after the program ends:

conn = sqlite3.connect(':memory:')

3.2 Create table

# 从连接获取游标
cursor = conn.cursor()

# 创建 users 表，IF NOT EXISTS 避免多次运行报错
cursor.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        email TEXT UNIQUE,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
''')

# 建表是写操作，记得提交
conn.commit()

3.3 Insert data

⚠️ **Important! Never use string concatenation to populate SQL parameters! ** Must be used?Parameterized queries for placeholders + tuples to prevent SQL injection.

Single insert

# 第二个参数必须是元组，哪怕只有一个值
cursor.execute(
    "INSERT INTO users (name, email) VALUES (?, ?)", 
    ('Alice', 'alice@example.com')
)
conn.commit()

Batch insert

executemany()Than loop callexecute()Much faster and recommended for batch writes:

new_users = [
    ('Bob', 'bob@example.com'),
    ('Charlie', 'charlie@example.com'),
    ('Diana', 'diana@example.com')
]

cursor.executemany(
    "INSERT INTO users (name, email) VALUES (?, ?)", 
    new_users
)
conn.commit()

3.4 Query data

There are three ways to obtain query results, choose as needed:

Get the full amount:fetchall()

cursor.execute("SELECT * FROM users")
all_users = cursor.fetchall()  # 返回一个列表，每个元素是一个元组

for user in all_users:
    print(user)

Single item acquisition:fetchone()

Suitable for querying unique results (such as querying by ID or unique key):

cursor.execute("SELECT name, email FROM users WHERE id = ?", (1,))
alice = cursor.fetchone()  # 返回一个元组，没查到则返回 None
print(alice)

Get in batches:fetchmany(n)

When querying large amounts of data, batch acquisition can avoid memory overflow caused by one-time loading:

cursor.execute("SELECT * FROM users")
while True:
    batch = cursor.fetchmany(2)  # 一次取 2 条
    if not batch:
        break
    for user in batch:
        print(user)

3.5 Update data

cursor.execute(
    "UPDATE users SET email = ? WHERE name = ?", 
    ('alice.updated@example.com', 'Alice')
)
conn.commit()

3.6 Delete data

cursor.execute("DELETE FROM users WHERE name = ?", ('Bob',))
conn.commit()

4. Advanced-features to improve development efficiency

4.1 Use context managers to automatically manage connections and transactions

Starting with Python 3.10,Connectionitself supportswithstatement, which has three benefits:

  • Exit normallywithblock, automatically commit the transaction
  • When an exception occurs, automatically roll back the transaction
  • After the block ends, automatically close the connection

Example:

with sqlite3.connect('example.db') as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    print(cursor.fetchall())

4.2 Use row factory to return dictionary

The default query returns tuples without field names, which is inconvenient in many scenarios. Can be set byrow_factoryPut the results into a dictionary or more readable format.

Custom dictionary factory

def dict_factory(cursor, row):
    return {col[0]: row[idx] for idx, col in enumerate(cursor.description)}

conn.row_factory = dict_factory
cursor = conn.cursor()

cursor.execute("SELECT * FROM users WHERE id = ?", (1,))
alice_dict = cursor.fetchone()
print(alice_dict["email"])  # 直接通过字段名访问

Use the built-insqlite3.Row

If you don’t want to write your own factory, Python’s built-insqlite3.RowMore lightweight, also supports field name access:

conn.row_factory = sqlite3.Row
cursor = conn.cursor()

cursor.execute("SELECT * FROM users WHERE id = ?", (1,))
alice_row = cursor.fetchone()
print(alice_row["email"])   # 用字段名访问
print(tuple(alice_row))     # 也可以转成普通元组

4.3 Explicit control of transactions

Most of the time usewithIt is enough, but submission and rollback can also be manually controlled under complex business logic:

try:
    cursor.execute("UPDATE users SET stock = stock - 1 WHERE id = ?", (1,))
    cursor.execute("INSERT INTO orders (user_id, product_id) VALUES (?, ?)", (1, 1))
    
    conn.commit()  # 全部成功后提交
except sqlite3.Error as e:
    print(f"操作失败，已回滚：{e}")
    conn.rollback()  # 任意一个失败就回滚

5. Best Practice Checklist

  1. ✅ Always use parameterized queries: Reject string concatenation SQL.
  2. ✅ UsewithStatement management connection: Automatically handle submission, rollback and closing to avoid resource leaks.
  3. ✅ Batch operations are given priorityexecutemany(): Performance improvement is obvious.
  4. ✅ Reasonable use of memory database: temporary testing or direct use of cache:memory:。
  5. ✅ Capturesqlite3.Error: Handle possible SQL syntax errors or constraint conflicts.
  6. ✅ Consider ORM for complex applications: For example, SQLAlchemy can map tables into Python classes to reduce handwritten SQL.

6. Complete runnable example: simple product inventory system

import sqlite3
from contextlib import closing  # 更规范地管理游标

def init_inventory_db():
    """初始化商品库存表"""
    with sqlite3.connect('inventory.db') as conn:
        with closing(conn.cursor()) as cursor:
            cursor.execute('''
                CREATE TABLE IF NOT EXISTS products (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    name TEXT NOT NULL UNIQUE,
                    price REAL NOT NULL CHECK (price > 0),
                    stock INTEGER DEFAULT 0 CHECK (stock >= 0)
                )
            ''')

def add_product(name: str, price: float, stock: int = 0) -> int | None:
    """添加商品，返回商品ID，失败返回None"""
    try:
        with sqlite3.connect('inventory.db') as conn:
            with closing(conn.cursor()) as cursor:
                cursor.execute(
                    "INSERT INTO products (name, price, stock) VALUES (?, ?, ?)",
                    (name, price, stock)
                )
                return cursor.lastrowid  # 获取刚插入的自增ID
    except sqlite3.IntegrityError as e:
        print(f"商品名重复或价格/库存非法：{e}")
        return None
    except sqlite3.Error as e:
        print(f"数据库异常：{e}")
        return None

def get_products_by_min_price(min_price: float = 0) -> list[dict]:
    """查询价格≥min_price的商品，返回字典列表，按价格升序"""
    try:
        with sqlite3.connect('inventory.db') as conn:
            conn.row_factory = sqlite3.Row
            with closing(conn.cursor()) as cursor:
                cursor.execute(
                    "SELECT * FROM products WHERE price >= ? ORDER BY price ASC",
                    (min_price,)
                )
                return [dict(row) for row in cursor.fetchall()]
    except sqlite3.Error as e:
        print(f"查询失败：{e}")
        return []

# 主程序示例
if __name__ == '__main__':
    init_inventory_db()
    
    # 添加商品
    add_product("机械键盘", 299.99, 50)
    add_product("无线鼠标", 89.99, 120)
    add_product("游戏耳机", 499.99, 30)
    
    # 查询价格≥200的商品
    premium_products = get_products_by_min_price(200)
    print("高端外设商品：")
    for p in premium_products:
        print(f"- {p['name']}: ¥{p['price']:.2f} (库存: {p['stock']})")

7. Supporting exercises and solutions

Exercise: Query students in a specified score range

Requirements: Implementationget_score_in(low, high)Function that returns the fraction in[low, high]The names of the students between, sorted by score from low to high.

Complete solution

import os
import sqlite3
from contextlib import closing

def get_score_in(low: int, high: int, db_path: str = 'test_scores.db') -> list[str]:
    with sqlite3.connect(db_path) as conn:
        with closing(conn.cursor()) as cursor:
            cursor.execute(
                "SELECT name FROM user WHERE score BETWEEN ? AND ? ORDER BY score ASC",
                (low, high)
            )
            return [row[0] for row in cursor.fetchall()]

# 测试代码
if __name__ == '__main__':
    TEST_DB = 'test_scores.db'
    
    # 清理旧测试数据库
    if os.path.exists(TEST_DB):
        os.remove(TEST_DB)
    
    # 初始化测试数据
    with sqlite3.connect(TEST_DB) as conn:
        with closing(conn.cursor()) as cursor:
            cursor.execute('''
                CREATE TABLE user (
                    id VARCHAR(20) PRIMARY KEY,
                    name VARCHAR(20) NOT NULL,
                    score INT NOT NULL CHECK (score BETWEEN 0 AND 100)
                )
            ''')
            cursor.executemany(
                "INSERT INTO user VALUES (?, ?, ?)",
                [
                    ('A-001', 'Adam', 95),
                    ('A-002', 'Bart', 62),
                    ('A-003', 'Lisa', 78)
                ]
            )
    
    # 执行测试
    assert get_score_in(80, 95) == ['Adam'], "80-95分测试失败"
    assert get_score_in(60, 80) == ['Bart', 'Lisa'], "60-80分测试失败"
    assert get_score_in(60, 100) == ['Bart', 'Lisa', 'Adam'], "60-100分测试失败"
    print("🎉 所有测试通过！")