当前位置：首页 > news >正文

【Flask】四、flask连接并操作数据库

news 2026/3/27 1:12:12

前言

一、?安装必要的库

二、配置数据库连接

?三、定义模型

四、操作数据库

1.添加用户

2.删除用户

3.更新用户信息

4查询所有用户

五、测试结果

前言

在Flask框架中，数据库的操作是一个核心功能，它允许开发者与后端数据库进行交互，执行数据的增删改查操作。本文将详细介绍如何在Flask中连接并操作数据库，包括使用Flask-SQLAlchemy扩展和直接使用MySQL的Python驱动。

常见的有三种操作方法：使用 SQLAlchemy、使用 Flask-SQLAlchemy 连接 MySQL和直接使用 MySQL 的 Python 驱动，本篇先讲述如何使用SQLAlchemy进行连接，后续会对其他方式进行讲解。

一、安装必要的库

首先，安装Flask-SQLAlchemy和MySQL的Python驱动。可以通过以下命令安装：

pip install -U SQLAlchemy pip install pymysql # Python3 使用pymysql pip install mysqlclient

二、配置数据库连接

在Flask应用中，需要配置数据库的地址（一般情况下，不建议直接在项目头部配置，此处只为演示，建议写在config.py中进行变量配置），这里的username等信息需要换成实际信息：

from flask import Flask from flask_sqlalchemy import SQLAlchemy app = Flask(__name__) app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://username:password@localhost/dbname' app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False db = SQLAlchemy(app)

三、定义模型

在Flask-SQLAlchemy中，模型是数据库表的Python类，每个模型类代表数据库中的一张表。例如：

class User(db.Model): __tablename__ = 'users' id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(80), unique=True, nullable=False) email = db.Column(db.String(120), unique=True, nullable=False) def __repr__(self): return f'<User {self.username}>'

这里定义了一个用户模型，对应数据库中的用户信息表，属性包括：id、用户名和email，对应的mysql表如下：

CREATE TABLE `users` ( `id` INT NOT NULL AUTO_INCREMENT, `username` VARCHAR(80) NOT NULL, `email` VARCHAR(120) NOT NULL, PRIMARY KEY (`id`), UNIQUE INDEX `username_unique` (`username` ASC), UNIQUE INDEX `email_unique` (`email` ASC) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

四、操作数据库

配置好以上内容后，我们就可以通过SQLAlchemy 提供的 ORM 功能操作数据库了，我们简单的用CURD操作进行测试~

1.添加用户

# 添加用户路由 @app.route('/add_user', methods=['POST']) def add_user(): data = request.get_json() username = data.get('username') email = data.get('email') if not username or not email: return '缺少用户名或邮箱地址', 400 # 检查 email 是否已存在 existing_user = User.query.filter_by(email=email).first() if existing_user: return '邮箱已存在，无法添加用户', 400 new_user = User(username=username, email=email) db.session.add(new_user) try: db.session.commit() except IntegrityError: db.session.rollback() return '插入失败，可能是用户名或邮箱重复', 400 except Exception as e: db.session.rollback() return f'数据库错误: {str(e)}', 500 return '用户添加成功'

2.删除用户

# 删除用户路由 @app.route('/delete_user/<int:user_id>', methods=['DELETE']) def delete_user(user_id): user = User.query.get(user_id) if not user: return '用户不存在', 404 db.session.delete(user) db.session.commit() return '用户删除成功'

3.更新用户信息

# 更新用户路由 @app.route('/update_user/<int:user_id>', methods=['PUT']) def update_user(user_id): user = User.query.get(user_id) if not user: return '用户不存在', 404 data = request.get_json() new_username = data.get('username') new_email = data.get('email') # 更新用户名和邮箱 if new_username: user.username = new_username if new_email: existing_user = User.query.filter_by(email=new_email).first() if existing_user and existing_user.id != user_id: return '邮箱已存在，无法更新用户信息', 400 user.email = new_email try: db.session.commit() except IntegrityError: db.session.rollback() return '更新失败，可能是用户名或邮箱重复', 400 except Exception as e: db.session.rollback() return f'数据库错误: {str(e)}', 500 return '用户更新成功'

4查询所有用户

# 获取所有用户路由 @app.route('/get_all', methods=['GET']) def get_users(): users = User.query.all() return '<br>'.join([f'{user.username} ({user.email})' for user in users])