一步步详解使用 Flask 连接数据库进行增删改查操作

下面我将一步步详细讲解如何使用 Flask 连接数据库进行增删改查操作（使用 SQLite 作为示例数据库），代码包含详细注释：

准备工作

1. 安装必要库：

pip install flask flask-sqlalchemy

完整代码 (app.py)

from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemyapp = Flask(__name__)
# 配置SQLite数据库路径（当前目录下的test.db文件）
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False  # 关闭警告# 初始化数据库对象
db = SQLAlchemy(app)# 定义数据模型（对应数据库表）
class User(db.Model):id = db.Column(db.Integer, primary_key=True)  # 主键username = db.Column(db.String(80), unique=True, nullable=False)email = db.Column(db.String(120), nullable=False)def __repr__(self):return f'<User {self.username}>'# 创建数据库表（首次运行）
@app.before_first_request
def create_tables():db.create_all()# 添加用户 (Create)
@app.route('/user', methods=['POST'])
def add_user():data = request.json# 检查必要字段if 'username' not in data or 'email' not in data:return jsonify({'error': 'Missing required fields'}), 400new_user = User(username=data['username'], email=data['email'])try:db.session.add(new_user)  # 添加到会话db.session.commit()       # 提交到数据库return jsonify({'message': 'User created', 'id': new_user.id}), 201except:db.session.rollback()     # 出错时回滚return jsonify({'error': 'Username already exists'}), 400# 查询所有用户 (Read)
@app.route('/users', methods=['GET'])
def get_all_users():users = User.query.all()# 将用户对象列表转换为字典列表output = [{'id': u.id, 'username': u.username, 'email': u.email} for u in users]return jsonify({'users': output})# 查询单个用户 (Read)
@app.route('/user/<int:user_id>', methods=['GET'])
def get_user(user_id):user = User.query.get(user_id)if not user:return jsonify({'error': 'User not found'}), 404return jsonify({'id': user.id, 'username': user.username, 'email': user.email})# 更新用户 (Update)
@app.route('/user/<int:user_id>', methods=['PUT'])
def update_user(user_id):user = User.query.get(user_id)if not user:return jsonify({'error': 'User not found'}), 404data = request.json# 更新字段（如果提供）if 'email' in data:user.email = data['email']try:db.session.commit()return jsonify({'message': 'User updated'})except:db.session.rollback()return jsonify({'error': 'Update failed'}), 500# 删除用户 (Delete)
@app.route('/user/<int:user_id>', methods=['DELETE'])
def delete_user(user_id):user = User.query.get(user_id)if not user:return jsonify({'error': 'User not found'}), 404try:db.session.delete(user)db.session.commit()return jsonify({'message': 'User deleted'})except:db.session.rollback()return jsonify({'error': 'Delete failed'}), 500if __name__ == '__main__':app.run(debug=True)  # 调试模式

代码详解

1. 初始化设置

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test.db'
db = SQLAlchemy(app)

• 配置 SQLite 数据库路径（会在项目目录下创建 test.db 文件）
• SQLAlchemy 是 Flask 的 ORM（对象关系映射）工具，让我们可以用 Python 类操作数据库

2. 定义数据模型

class User(db.Model):id = db.Column(db.Integer, primary_key=True)username = db.Column(db.String(80), unique=True, nullable=False)email = db.Column(db.String(120), nullable=False)

• 每个类属性对应数据库表的列
• primary_key=True 表示主键
• unique=True 确保用户名唯一
• nullable=False 表示不能为空

3. 创建数据库表

@app.before_first_request
def create_tables():db.create_all()

• 在首次请求时自动创建表结构
• 实际项目中建议使用数据库迁移工具（如 Flask-Migrate）

4. CRUD 操作实现

Create (添加用户)

new_user = User(username='test', email='test@example.com')
db.session.add(new_user)
db.session.commit()

• session.add() 将对象加入会话
• session.commit() 提交事务到数据库

Read (查询用户)

# 查询所有用户
User.query.all()# 按ID查询
User.query.get(1)# 条件查询
User.query.filter_by(username='john').first()

Update (更新用户)

user = User.query.get(1)
user.email = 'new@example.com'
db.session.commit()

Delete (删除用户)

user = User.query.get(1)
db.session.delete(user)
db.session.commit()

使用 Postman 测试 API

1. 添加用户 (POST)

   URL: http://localhost:5000/user
   Body (JSON):
   {"username": "john","email": "john@example.com"
   }
   

2. 查询所有用户 (GET)

   URL: http://localhost:5000/users
   

3. 查询单个用户 (GET)

   URL: http://localhost:5000/user/1
   

4. 更新用户 (PUT)

   URL: http://localhost:5000/user/1
   Body (JSON):
   {"email": "updated@example.com"
   }
   

5. 删除用户 (DELETE)

   URL: http://localhost:5000/user/1
   

关键注意事项

1. 错误处理：

   • 所有数据库操作都使用 try-except 处理异常
   • 出错时执行 rollback() 回滚事务

2. 会话管理：

   • 每次数据库操作后必须 commit() 才会生效
   • 相关操作应在同一个会话中完成

3. 数据验证：

   • 实际项目中应添加更严格的数据验证
   • 推荐使用 Flask-WTF 或 marshmallow

4. 连接其他数据库：

   • MySQL: mysql://username:password@localhost/dbname
   • PostgreSQL: postgresql://username:password@localhost/dbname

5. 生产环境优化：

   • 关闭调试模式 (debug=False)
   • 使用数据库连接池
   • 添加身份验证

这个示例提供了完整的 RESTful API 实现，包含了数据库操作的基本模式。实际开发中可以根据需求扩展功能，如添加分页、搜索过滤、关联关系等。