数据库支持13.1 数据库概述13.1.1 为什么需要数据库当程序需要存储和管理大量结构化数据时普通文件如 CSV、JSON存在明显局限13.1.2 Python 数据库 APIDB APIPython 提供了统一的数据库接口规范DB API 2.0PEP 249不同数据库的 Python 模块遵循相同的 API让代码可以更容易地切换数据库。# 统一的使用模式不因数据库类型而变 import sqlite3 # 或 psycopg2, MySQLdb 等 conn connect(database.db) # 连接数据库 curs conn.cursor() # 获取游标 curs.execute(SELECT * FROM table) # 执行 SQL rows curs.fetchall() # 获取结果 conn.commit() # 提交事务 conn.close() # 关闭连接13.1.3 常见数据库13.2 SQLite 快速入门13.2.1 连接数据库import sqlite3 # 连接到数据库文件不存在时会自动创建 conn sqlite3.connect(mydatabase.db) # 连接内存数据库临时程序结束即消失 conn sqlite3.connect(:memory:) # 关闭连接 conn.close()13.2.2 游标Cursor游标用于执行 SQL 语句和获取查询结果。import sqlite3 conn sqlite3.connect(mydatabase.db) curs conn.cursor() # 执行 SQL 语句 curs.execute(CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)) # 获取结果 rows curs.fetchall() conn.close()13.3 创建表和插入数据13.3.1 创建表import sqlite3 conn sqlite3.connect(mydatabase.db) curs conn.cursor() # 创建用户表 curs.execute( CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, age INTEGER, email TEXT UNIQUE ) ) conn.commit() # 提交更改 conn.close()SQLite 数据类型常用约束13.3.2 插入数据三种方式方式1直接拼接字符串不安全不推荐# 危险容易遭受 SQL 注入攻击 name Alice; DROP TABLE users; -- curs.execute(fINSERT INTO users (name) VALUES ({name}))方式2使用参数化查询推荐# 使用 ? 占位符 name Alice age 25 curs.execute(INSERT INTO users (name, age) VALUES (?, ?), (name, age)) # 使用命名占位符 curs.execute(INSERT INTO users (name, age) VALUES (:name, :age), {name: Bob, age: 30})方式3批量插入users [ (Charlie, 35), (Diana, 28), (Eve, 42), ] curs.executemany(INSERT INTO users (name, age) VALUES (?, ?), users)13.3.3 提交事务conn.commit() # 提交所有更改 # 如果不提交关闭连接时更改会丢失13.4 查询数据13.4.1 基本查询import sqlite3 conn sqlite3.connect(mydatabase.db) curs conn.cursor() # 执行查询 curs.execute(SELECT * FROM users) # 获取所有行返回列表 all_rows curs.fetchall() print(all_rows) # [(1, Alice, 25, None), (2, Bob, 30, None), ...] # 获取一行 one_row curs.fetchone() # 获取多行指定数量 few_rows curs.fetchmany(5) conn.close()13.4.2 参数化查询# 查询特定年龄的用户 curs.execute(SELECT * FROM users WHERE age ?, (25,)) # 查询名字包含特定字符的用户 curs.execute(SELECT * FROM users WHERE name LIKE ?, (A%,)) # 使用命名参数 curs.execute(SELECT * FROM users WHERE name :name, {name: Alice})13.4.3 使用字典游标默认返回元组可以使用字典游标获取带列名的结果。conn sqlite3.connect(mydatabase.db) conn.row_factory sqlite3.Row # 设置为 Row 工厂 curs conn.cursor() curs.execute(SELECT * FROM users) # 结果可以像字典一样访问 for row in curs.fetchall(): print(row[name], row[age]) # 或者转换为普通字典 rows [dict(row) for row in curs.fetchall()]13.4.4 获取列名curs.execute(SELECT * FROM users) col_names [description[0] for description in curs.description] print(col_names) # [id, name, age, email]13.5 更新和删除13.5.1 更新数据# 更新特定用户 curs.execute(UPDATE users SET age ? WHERE name ?, (26, Alice)) # 批量更新所有用户年龄加1 curs.execute(UPDATE users SET age age 1) conn.commit()13.5.2 删除数据# 删除特定用户 curs.execute(DELETE FROM users WHERE name ?, (Eve,)) # 删除所有数据保留表结构 curs.execute(DELETE FROM users) conn.commit()13.6 事务处理13.6.1 事务的概念事务是一组数据库操作的集合要么全部成功提交要么全部失败回滚。13.6.2 事务示例import sqlite3 conn sqlite3.connect(bank.db) curs conn.cursor() try: # 开始事务自动开始 curs.execute(UPDATE accounts SET balance balance - 100 WHERE id 1) curs.execute(UPDATE accounts SET balance balance 100 WHERE id 2) # 提交事务 conn.commit() print(转账成功) except Exception as e: # 发生错误回滚所有更改 conn.rollback() print(转账失败已回滚:, e) finally: conn.close()13.6.3 自动提交模式# 默认情况下sqlite3 使用隐式事务 # 可以启用自动提交每个操作立即生效 conn sqlite3.connect(mydatabase.db, isolation_levelNone)13.7 综合示例营养成分数据库下面是一个从数据文件导入到 SQLite 数据库并进行查询的完整示例。13.7.1 数据结构假设有一个名为ABBREV.txt的数据文件字段用^分隔字符串字段用~包围。~07276~^~HORMEL SPAM...PORK W/ HAM MINCED CND~^...^~1 serving~^^~~^013.7.2 导入数据import sqlite3 def convert(value): 转换字段值 if value.startswith(~): return value.strip(~) # 字符串 if not value: return 0 # 空值转为0 return float(value) # 数字 def import_data(filenameABBREV.txt): conn sqlite3.connect(food.db) curs conn.cursor() # 创建表 curs.execute( CREATE TABLE IF NOT EXISTS food ( id TEXT PRIMARY KEY, desc TEXT, water REAL, kcal REAL, protein REAL, fat REAL, ash REAL, carbs REAL, fiber REAL, sugar REAL ) ) # 插入数据 field_count 10 query INSERT INTO food VALUES ( ,.join([?]*10) ) with open(filename, r, encodingutf-8) as f: for line in f: fields line.split(^) vals [convert(f) for f in fields[:field_count]] curs.execute(query, vals) conn.commit() conn.close() print(f数据导入完成)13.7.3 查询数据def query_food(condition): 根据条件查询食品数据 conn sqlite3.connect(food.db) conn.row_factory sqlite3.Row curs conn.cursor() query fSELECT * FROM food WHERE {condition} curs.execute(query) # 获取列名 col_names [desc[0] for desc in curs.description] # 打印结果 for row in curs.fetchall(): for name in col_names: print(f{name}: {row[name]}) print() conn.close() # 使用示例查找低热量高纤维的食品 query_food(kcal 100 AND fiber 10 ORDER BY sugar)13.8 SQL 注入防护重要13.8.1 什么是 SQL 注入SQL 注入是攻击者通过构造恶意输入来执行非预期的 SQL 语句从而破坏数据库的攻击方式。# 危险代码 user_input Alice; DROP TABLE users; -- curs.execute(fSELECT * FROM users WHERE name {user_input}) # 实际执行SELECT * FROM users WHERE name Alice; DROP TABLE users; -- # ↑ 恶意代码被注入并执行13.8.2 防护方法永远使用参数化查询# 安全方式使用参数化查询 user_input Alice; DROP TABLE users; -- curs.execute(SELECT * FROM users WHERE name ?, (user_input,)) # 参数被当作普通字符串处理不会执行恶意代码13.9 本章知识点速查表