Flask如何查询MySQL数据库？Flask连接MySQL实现增删改查

在 Flask 中查询 MySQL 数据库，通常有以下几种主流方式：

1. Flask-SQLAlchemy（推荐，ORM 方式，适合大多数项目）
2. PyMySQL + Flask（原生 SQL 方式，轻量级）
3. SQLAlchemy + Flask（不使用 Flask-SQLAlchemy 扩展，更灵活）

下面我将分别介绍这三种方式,并提供完整示例。

加载中

python flask连接mysql数据库实现登录注册

ns阿熏

5.5万81529

原视频地址

✅ 方式一：使用 Flask-SQLAlchemy（推荐）

安装依赖

pip install Flask Flask-SQLAlchemy pymysql

配置数据库连接

在 app.py 中：

from flask import Flask, jsonify
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
# 配置 MySQL 数据库连接
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://username:password@localhost:3306/your_database'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# 初始化 SQLAlchemy
db = SQLAlchemy(app)
# 定义模型（表结构）
class User(db.Model):
    __tablename__ = 'users'
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50), nullable=False)
    email = db.Column(db.String(100), unique=True, nullable=False)
    def to_dict(self):
        return {
            'id': self.id,
            'name': self.name,
            'email': self.email
        }
# 创建表（首次运行）
with app.app_context():
    db.create_all()
# 查询所有用户
@app.route('/users', methods=['GET'])
def get_users():
    users = User.query.all()
    return jsonify([user.to_dict() for user in users])
# 根据 ID 查询用户
@app.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    user = User.query.get(user_id)
    if user:
        return jsonify(user.to_dict())
    else:
        return jsonify({'error': 'User not found'}), 404
if __name__ == '__main__':
    app.run(debug=True)

常用查询示例

# 查询所有
users = User.query.all()
# 查询第一个
first_user = User.query.first()
# 根据 ID 查询
user = User.query.get(1)
# 条件查询
users = User.query.filter_by(name='Alice').all()
users = User.query.filter(User.email.like('%@gmail.com')).all()
# 分页查询
users = User.query.paginate(page=1, per_page=10)
# 统计数量
count = User.query.count()
# 复杂查询
users = User.query.filter(User.name.like('%a%')).order_by(User.id.desc()).all()

✅ 方式二：使用 PyMySQL + Flask（原生 SQL）

安装依赖

pip install Flask pymysql

示例代码

from flask import Flask, jsonify
import pymysql
app = Flask(__name__)
# 数据库配置
DB_CONFIG = {
    'host': 'localhost',
    'user': 'username',
    'password': 'password',
    'database': 'your_database',
    'charset': 'utf8mb4',
    'cursorclass': pymysql.cursors.DictCursor
}
# 获取数据库连接
def get_db_connection():
    return pymysql.connect(DB_CONFIG)
# 查询所有用户
@app.route('/users', methods=['GET'])
def get_users():
    conn = get_db_connection()
    try:
        with conn.cursor() as cursor:
            sql = "SELECT id, name, email FROM users"
            cursor.execute(sql)
            users = cursor.fetchall()
        return jsonify(users)
    finally
:
        conn.close()
# 根据 ID 查询用户
@app.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    conn = get_db_connection()
    try:
        with conn.cursor() as cursor:
            sql = "SELECT id, name, email FROM users WHERE id = %s"
            cursor.execute(sql, (user_id,))
            user = cursor.fetchone()
        if user:
            return jsonify(user)
        else:
            return jsonify({'error': 'User not found'}), 404
    finally:
        conn.close()
if __name__ == '__main__':
    app.run(debug=True)

✅ 方式三：使用 SQLAlchemy + Flask（不使用 Flask-SQLAlchemy）

安装依赖

pip install Flask SQLAlchemy pymysql

示例代码

from flask import Flask, jsonify
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
app = Flask(__name__)
# 数据库引擎
engine = create_engine('mysql+pymysql://username:password@localhost:3306/your_database')
# 创建基类
Base = declarative_base()
# 定义模型
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, nullable=False)
# 创建会话工厂
Session = sessionmaker(bind=engine)
# 创建表
Base.metadata.create_all(engine)
# 查询所有用户
@app.route('/users', methods=['GET'])
def get_users():
    session = Session()
    try:
        users = session.query(User).all()
        return jsonify([
            {'id': user.id, 'name': user.name, 'email': use
r.email}
            for user in users
        ])
    finally:
        session.close()
# 根据 ID 查询用户
@app.route('/users/<int:user_id>', methods=['GET'])
def get_user(user_id):
    session = Session()
    try:
        user = session.query(User).filter_by(id=user_id).first()
        if user:
            return jsonify({
                'id': user.id,
                'name': user.name,
                'email': user.email
            })
        else:
            return jsonify({'error': 'User not found'}), 404
    finally:
        session.close()
if __name__ == '__main__':
    app.run(debug=True)

📌 总结对比

⚠️ 安全提示

1. 防止 SQL 注入：使用参数化查询（如 cursor.execute(sql, (param,)) 或 ORM 的 filter_by）
2. 使用环境变量：数据库密码不要硬编码，使用 os.environ 或 .env 文件
3. 连接池：生产环境建议使用连接池（如 SQLAlchemy 自带的连接池）