《Python开发从入门到精通》设计指南第四篇：数据库编程与数据存储
一、学习目标与重点
💡 学习目标：掌握Python数据库编程的基本方法，包括MySQL、SQLite和MongoDB的连接、查询、插入、更新和删除操作；理解ORM（对象关系映射）的概念，掌握SQLAlchemy库的使用。 ⚠️ 学习重点：SQL语句基础、数据库连接配置、CRUD操作、事务管理、ORM映射、MongoDB操作。
4.1 数据库概述
4.1.1 数据库类型
- 关系型数据库：如MySQL、PostgreSQL、SQLite、Oracle等，数据以表的形式存储，使用SQL（Structured Query Language）进行操作。
- 非关系型数据库：如MongoDB、Redis、Cassandra等，数据以键值对、文档、列族等形式存储，不使用SQL。
4.1.2 常用数据库
- MySQL：开源的关系型数据库，广泛应用于Web开发。
- SQLite：轻量级的关系型数据库，数据存储在单个文件中，适合小型应用。
- MongoDB：开源的文档型数据库，数据以BSON（Binary JSON）格式存储。
4.2 SQLite数据库操作
4.2.1 SQLite简介
SQLite是Python标准库中自带的数据库，无需额外安装，使用简单。
4.2.2 连接数据库
import sqlite3
# 连接到SQLite数据库（如果不存在则创建）
conn = sqlite3.connect('example.db')
# 创建游标对象
cursor = conn.cursor()
# 执行SQL语句
cursor.execute('''CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
age INTEGER NOT NULL,
email TEXT UNIQUE NOT NULL)''')
# 提交事务
conn.commit()
# 关闭连接
conn.close()
4.2.3 CRUD操作
4.2.3.1 插入数据
import sqlite3
conn = sqlite3.connect('example.db')
cursor = conn.cursor()
# 插入单条数据
cursor.execute('''INSERT INTO users (name, age, email) VALUES (?, ?, ?)''', ('张三', 25, 'zhangsan@example.com'))
# 插入多条数据
users = [
('李四', 26, 'lisi@example.com'),
('王五', 24, 'wangwu@example.com')
]
cursor.executemany('''INSERT INTO users (name, age, email) VALUES (?, ?, ?)''', users)
conn.commit()
conn.close()
4.2.3.2 查询数据
import sqlite3
conn = sqlite3.connect('example.db')
cursor = conn.cursor()
# 查询所有数据
cursor.execute('''SELECT * FROM users''')
all_users = cursor.fetchall()
for user in all_users:
print(user)
# 查询单条数据
cursor.execute('''SELECT * FROM users WHERE id = ?''', (1,))
user = cursor.fetchone()
print(user)
# 查询指定列的数据
cursor.execute('''SELECT name, email FROM users WHERE age > ?''', (25,))
users = cursor.fetchall()
for user in users:
print(user)
conn.close()
4.2.3.3 更新数据
import sqlite3
conn = sqlite3.connect('example.db')
cursor = conn.cursor()
cursor.execute('''UPDATE users SET age = ? WHERE name = ?''', (27, '李四'))
conn.commit()
conn.close()
4.2.3.4 删除数据
import sqlite3
conn = sqlite3.connect('example.db')
cursor = conn.cursor()
cursor.execute('''DELETE FROM users WHERE id = ?''', (3,))
conn.commit()
conn.close()
4.3 MySQL数据库操作
4.3.1 安装MySQL连接器
pip install mysql-connector-python
4.3.2 连接数据库
import mysql.connector
# 连接到MySQL数据库
conn = mysql.connector.connect(
host='localhost',
user='root',
password='password',
database='mydatabase'
)
# 创建游标对象
cursor = conn.cursor()
# 执行SQL语句
cursor.execute('''CREATE TABLE IF NOT EXISTS users (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(255) NOT NULL,
age INT NOT NULL,
email VARCHAR(255) UNIQUE NOT NULL)''')
# 提交事务
conn.commit()
# 关闭连接
conn.close()
4.3.3 CRUD操作
4.3.3.1 插入数据
import mysql.connector
conn = mysql.connector.connect(
host='localhost',
user='root',
password='password',
database='mydatabase'
)
cursor = conn.cursor()
# 插入单条数据
sql = '''INSERT INTO users (name, age, email) VALUES (%s, %s, %s)'''
val = ('张三', 25, 'zhangsan@example.com')
cursor.execute(sql, val)
# 插入多条数据
sql = '''INSERT INTO users (name, age, email) VALUES (%s, %s, %s)'''
vals = [
('李四', 26, 'lisi@example.com'),
('王五', 24, 'wangwu@example.com')
]
cursor.executemany(sql, vals)
conn.commit()
print(f"插入了{cursor.rowcount}条数据")
conn.close()
4.3.3.2 查询数据
import mysql.connector
conn = mysql.connector.connect(
host='localhost',
user='root',
password='password',
database='mydatabase'
)
cursor = conn.cursor()
# 查询所有数据
cursor.execute('''SELECT * FROM users''')
all_users = cursor.fetchall()
for user in all_users:
print(user)
# 查询单条数据
cursor.execute('''SELECT * FROM users WHERE id = %s''', (1,))
user = cursor.fetchone()
print(user)
# 查询指定列的数据
cursor.execute('''SELECT name, email FROM users WHERE age > %s''', (25,))
users = cursor.fetchall()
for user in users:
print(user)
conn.close()
4.3.3.3 更新数据
import mysql.connector
conn = mysql.connector.connect(
host='localhost',
user='root',
password='password',
database='mydatabase'
)
cursor = conn.cursor()
sql = '''UPDATE users SET age = %s WHERE name = %s'''
val = (27, '李四')
cursor.execute(sql, val)
conn.commit()
print(f"更新了{cursor.rowcount}条数据")
conn.close()
4.3.3.4 删除数据
import mysql.connector
conn = mysql.connector.connect(
host='localhost',
user='root',
password='password',
database='mydatabase'
)
cursor = conn.cursor()
sql = '''DELETE FROM users WHERE id = %s'''
val = (3,)
cursor.execute(sql, val)
conn.commit()
print(f"删除了{cursor.rowcount}条数据")
conn.close()
4.4 MongoDB数据库操作
4.4.1 安装PyMongo
pip install pymongo
4.4.2 连接数据库
from pymongo import MongoClient
# 连接到MongoDB服务器
client = MongoClient('mongodb://localhost:27017/')
# 创建或选择数据库
db = client['mydatabase']
# 创建或选择集合
collection = db['users']
4.4.3 CRUD操作
4.4.3.1 插入数据
from pymongo import MongoClient
client = MongoClient('mongodb://localhost:27017/')
db = client['mydatabase']
collection = db['users']
# 插入单条数据
user = {
'name': '张三',
'age': 25,
'email': 'zhangsan@example.com'
}
result = collection.insert_one(user)
print(f"插入了文档，ID：{result.inserted_id}")
# 插入多条数据
users = [
{'name': '李四', 'age': 26, 'email': 'lisi@example.com'},
{'name': '王五', 'age': 24, 'email': 'wangwu@example.com'}
]
result = collection.insert_many(users)
print(f"插入了{len(result.inserted_ids)}条文档，ID：{result.inserted_ids}")
4.4.3.2 查询数据
from pymongo import MongoClient
client = MongoClient('mongodb://localhost:27017/')
db = client['mydatabase']
collection = db['users']
# 查询所有数据
all_users = collection.find()
for user in all_users:
print(user)
# 查询单条数据
user = collection.find_one({'name': '张三'})
print(user)
# 查询指定条件的数据
users = collection.find({'age': {'$gt': 25}})
for user in users:
print(user)
# 查询指定列的数据
users = collection.find({}, {'name': 1, 'email': 1})
for user in users:
print(user)
4.4.3.3 更新数据
from pymongo import MongoClient
client = MongoClient('mongodb://localhost:27017/')
db = client['mydatabase']
collection = db['users']
# 更新单条数据
result = collection.update_one({'name': '李四'}, {'$set': {'age': 27}})
print(f"匹配了{result.matched_count}条文档，修改了{result.modified_count}条文档")
# 更新多条数据
result = collection.update_many({'age': {'$lt': 25}}, {'$inc': {'age': 1}})
print(f"匹配了{result.matched_count}条文档，修改了{result.modified_count}条文档")
4.4.3.4 删除数据
from pymongo import MongoClient
client = MongoClient('mongodb://localhost:27017/')
db = client['mydatabase']
collection = db['users']
# 删除单条数据
result = collection.delete_one({'name': '王五'})
print(f"删除了{result.deleted_count}条文档")
# 删除多条数据
result = collection.delete_many({'age': {'$gt': 26}})
print(f"删除了{result.deleted_count}条文档")
4.5 ORM与SQLAlchemy
4.5.1 安装SQLAlchemy
pip install sqlalchemy
4.5.2 定义数据模型
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 创建引擎
engine = create_engine('sqlite:///example.db', echo=True)
# 创建基类
Base = declarative_base()
# 定义数据模型
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String, nullable=False)
age = Column(Integer, nullable=False)
email = Column(String, unique=True, nullable=False)
def __repr__(self):
return f"<User(name='{self.name}', age={self.age}, email='{self.email}')>"
# 创建表
Base.metadata.create_all(engine)
# 创建会话
Session = sessionmaker(bind=engine)
session = Session()
4.5.3 CRUD操作
4.5.3.1 插入数据
from models import User, session
# 插入单条数据
user1 = User(name='张三', age=25, email='zhangsan@example.com')
session.add(user1)
# 插入多条数据
user2 = User(name='李四', age=26, email='lisi@example.com')
user3 = User(name='王五', age=24, email='wangwu@example.com')
session.add_all([user2, user3])
session.commit()
4.5.3.2 查询数据
from models import User, session
# 查询所有数据
all_users = session.query(User).all()
for user in all_users:
print(user)
# 查询单条数据
user = session.query(User).filter_by(name='张三').first()
print(user)
# 查询指定条件的数据
users = session.query(User).filter(User.age > 25).all()
for user in users:
print(user)
# 查询指定列的数据
users = session.query(User.name, User.email).all()
for user in users:
print(user)
4.5.3.3 更新数据
from models import User, session
user = session.query(User).filter_by(name='李四').first()
if user:
user.age = 27
session.commit()
print("更新成功")
else:
print("未找到用户")
4.5.3.4 删除数据
from models import User, session
user = session.query(User).filter_by(name='王五').first()
if user:
session.delete(user)
session.commit()
print("删除成功")
else:
print("未找到用户")
4.6 实战案例：学生信息管理系统
4.6.1 需求分析
设计一个学生信息管理系统，支持以下功能：
- 添加学生信息（姓名、年龄、学号、班级）。
- 删除学生信息（学号）。
- 修改学生信息（学号）。
- 查询学生信息（学号）。
- 显示所有学生信息。
- 将学生信息存储到MySQL数据库中。
4.6.2 代码实现
import mysql.connector
from mysql.connector import Error
class StudentManagementSystem:
def __init__(self, host, user, password, database):
self.host = host
self.user = user
self.password = password
self.database = database
self.conn = None
self.cursor = None
def connect(self):
try:
self.conn = mysql.connector.connect(
host=self.host,
user=self.user,
password=self.password,
database=self.database
)
if self.conn.is_connected():
print("连接到MySQL数据库成功")
self.cursor = self.conn.cursor()
except Error as e:
print(f"连接到MySQL数据库失败：{e}")
def create_table(self):
try:
sql = '''CREATE TABLE IF NOT EXISTS students (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(255) NOT NULL,
age INT NOT NULL,
student_id VARCHAR(255) UNIQUE NOT NULL,
class_name VARCHAR(255) NOT NULL)'''
self.cursor.execute(sql)
self.conn.commit()
print("学生表创建成功")
except Error as e:
print(f"创建学生表失败：{e}")
def add_student(self, name, age, student_id, class_name):
try:
sql = '''INSERT INTO students (name, age, student_id, class_name) VALUES (%s, %s, %s, %s)'''
val = (name, age, student_id, class_name)
self.cursor.execute(sql, val)
self.conn.commit()
print(f"添加学生信息成功，学号：{student_id}")
except Error as e:
print(f"添加学生信息失败：{e}")
def delete_student(self, student_id):
try:
sql = '''DELETE FROM students WHERE student_id = %s'''
val = (student_id,)
self.cursor.execute(sql, val)
self.conn.commit()
print(f"删除学生信息成功，学号：{student_id}")
except Error as e:
print(f"删除学生信息失败：{e}")
def modify_student(self, student_id, name, age, class_name):
try:
sql = '''UPDATE students SET name = %s, age = %s, class_name = %s WHERE student_id = %s'''
val = (name, age, class_name, student_id)
self.cursor.execute(sql, val)
self.conn.commit()
print(f"修改学生信息成功，学号：{student_id}")
except Error as e:
print(f"修改学生信息失败：{e}")
def query_student(self, student_id):
try:
sql = '''SELECT * FROM students WHERE student_id = %s'''
val = (student_id,)
self.cursor.execute(sql, val)
student = self.cursor.fetchone()
if student:
print(f"学号：{student[3]}，姓名：{student[1]}，年龄：{student[2]}，班级：{student[4]}")
else:
print(f"未找到学号为{student_id}的学生")
except Error as e:
print(f"查询学生信息失败：{e}")
def display_all_students(self):
try:
sql = '''SELECT * FROM students'''
self.cursor.execute(sql)
students = self.cursor.fetchall()
if students:
print("所有学生信息：")
for student in students:
print(f"学号：{student[3]}，姓名：{student[1]}，年龄：{student[2]}，班级：{student[4]}")
else:
print("没有学生信息")
except Error as e:
print(f"显示学生信息失败：{e}")
def close(self):
if self.conn.is_connected():
self.cursor.close()
self.conn.close()
print("连接关闭")
def main():
sms = StudentManagementSystem(
host='localhost',
user='root',
password='password',
database='mydatabase'
)
sms.connect()
sms.create_table()
while True:
print("\\n学生信息管理系统")
print("1. 添加学生信息")
print("2. 删除学生信息")
print("3. 修改学生信息")
print("4. 查询学生信息")
print("5. 显示所有学生信息")
print("6. 退出系统")
choice = input("请输入你的选择（1/2/3/4/5/6）：")
if choice == "1":
name = input("请输入姓名：")
age = int(input("请输入年龄："))
student_id = input("请输入学号：")
class_name = input("请输入班级：")
sms.add_student(name, age, student_id, class_name)
elif choice == "2":
student_id = input("请输入学号：")
sms.delete_student(student_id)
elif choice == "3":
student_id = input("请输入学号：")
name = input("请输入新姓名：")
age = int(input("请输入新年龄："))
class_name = input("请输入新班级：")
sms.modify_student(student_id, name, age, class_name)
elif choice == "4":
student_id = input("请输入学号：")
sms.query_student(student_id)
elif choice == "5":
sms.display_all_students()
elif choice == "6":
print("感谢使用学生信息管理系统")
sms.close()
break
else:
print("无效的选择")
if __name__ == "__main__":
main()
4.6.3 实施过程
① 导入mysql.connector模块和Error异常类。 ② 定义StudentManagementSystem类，用于管理学生信息，实现连接、创建表、添加、删除、修改、查询和显示所有学生信息的功能。 ③ 定义main()函数，负责用户交互和系统调用。 ④ 运行程序，连接到MySQL数据库，创建学生表，并进行学生信息管理操作。
4.6.4 最终效果
运行程序后，用户可以根据菜单选择功能，输入相应的信息完成学生信息管理操作。程序会对无效输入和重复学号进行处理，并给出相应的提示。
4.7 常见问题与解决方案
4.7.1 数据库连接失败
问题：连接到数据库时提示Error，如权限不足、主机不可达等。 解决方案： ① 检查数据库服务器是否正在运行。 ② 检查连接参数是否正确（如用户名、密码、主机名、端口号等）。 ③ 确保防火墙允许该端口的入站连接。
4.7.2 SQL语句错误
问题：执行SQL语句时提示ProgrammingError，如语法错误、列名不存在等。 解决方案： ① 检查SQL语句的语法是否正确。 ② 检查表名、列名是否与数据库中的定义一致。 ③ 使用参数化查询代替字符串拼接，避免SQL注入攻击。
4.7.3 ORM操作错误
问题：使用SQLAlchemy时提示AttributeError或SQLAlchemyError，如属性未定义、关系未配置等。 解决方案： ① 检查数据模型的定义是否正确。 ② 检查会话是否正确创建和管理。 ③ 确保数据库中的表结构与数据模型一致。
总结
✅ 本文详细介绍了Python数据库编程的基本方法，包括MySQL、SQLite和MongoDB的连接、查询、插入、更新和删除操作；理解了ORM的概念，掌握了SQLAlchemy库的使用。 ✅ 通过学生信息管理系统的实战案例，读者可以理解数据库编程在实际开发中的应用。 ✅ 建议读者在学习过程中多练习，通过编写代码加深对知识点的理解。
