Python操作SQLite数据库增删改查完整教程

用Python内置sqlite3模块完成SQLite数据库的建表、插入、查询、更新、删除全流程，代码可直接运行，适合办公自动化数据落库场景。

场景痛点

做报表、跑脚本经常要把中间结果存到本地数据库，又不想装MySQL、PostgreSQL这种重型服务。SQLite是单文件数据库，零配置、随Python自带，适合存客户名单、订单明细、日志记录这类中小型数据。手动在DB Browser里一条条改太费时间，用Python脚本一次跑通增删改查，后续每天定时执行即可。

用到的库

sqlite3是Python标准库，无需pip安装。如果要配合pandas，再装：

pip install pandas

完整代码

import sqlite3
from pathlib import Path

DB_PATH = Path("office.db")

def get_conn():
"""获取数据库连接，数据库文件不存在会自动创建"""
conn = sqlite3.connect(DB_PATH)
# 让查询结果可以按列名取值，类似字典
conn.row_factory = sqlite3.Row
return conn

def create_table():
"""建表：员工信息表"""
conn = get_conn()
cur = conn.cursor()
cur.execute("""
CREATE TABLE IF NOT EXISTS employees (
id      INTEGER PRIMARY KEY AUTOINCREMENT,
name    TEXT NOT NULL,
dept    TEXT,
salary  REAL,
city    TEXT
)
""")
conn.commit()
conn.close()
print("建表完成")

def insert_data():
"""插入单条与批量插入"""
conn = get_conn()
cur = conn.cursor()
# 单条插入，用占位符 ? 防止SQL注入
cur.execute(
"INSERT INTO employees (name, dept, salary, city) VALUES (?, ?, ?, ?)",
("张三", "技术部", 15000.0, "北京")
)
# 批量插入
rows = [
("李四", "市场部", 12000.0, "上海"),
("王五", "技术部", 18000.0, "深圳"),
("赵六", "人事部", 9000.0, "北京"),
]
cur.executemany(
"INSERT INTO employees (name, dept, salary, city) VALUES (?, ?, ?, ?)",
rows
)
conn.commit()
print(f"插入完成，自增ID={cur.lastrowid}")
conn.close()

def query_data():
"""查询：全表、条件查询、排序、聚合"""
conn = get_conn()
cur = conn.cursor()

print("--- 全表查询 ---")
cur.execute("SELECT * FROM employees")
for row in cur.fetchall():
print(dict(row))

print("--- 条件查询：技术部员工 ---")
cur.execute("SELECT name, salary FROM employees WHERE dept = ?", ("技术部",))
for row in cur.fetchall():
print(row["name"], row["salary"])

print("--- 分组聚合：各部门平均工资 ---")
cur.execute("""
SELECT dept, COUNT(*) AS cnt, AVG(salary) AS avg_sal
FROM employees GROUP BY dept
""")
for row in cur.fetchall():
print(row["dept"], "人数:", row["cnt"], "平均:", round(row["avg_sal"], 2))

conn.close()

def update_data():
"""更新：按条件修改工资"""
conn = get_conn()
cur = conn.cursor()
cur.execute("UPDATE employees SET salary = ? WHERE name = ?", (16000.0, "张三"))
conn.commit()
print(f"更新 {cur.rowcount} 行")
conn.close()

def delete_data():
"""删除：按条件删除"""
conn = get_conn()
cur = conn.cursor()
cur.execute("DELETE FROM employees WHERE name = ?", ("赵六",))
conn.commit()
print(f"删除 {cur.rowcount} 行")
conn.close()

if __name__ == "__main__":
create_table()
insert_data()
query_data()
update_data()
delete_data()