Flask与MySQL数据库集成

用户8589624

发布于 2025-11-15 19:20:06

1280

文章被收录于专栏：nginxnginx

Flask与MySQL数据库集成

第一部分：MySQL 8安装与配置

1. MySQL 8安装指南

Windows平台安装

下载MySQL安装包

访问MySQL官网(https://dev.mysql.com/downloads/installer/)
下载MySQL Installer for Windows

安装步骤

# 1. 运行安装程序，选择"Developer Default"安装类型
# 2. 在安装过程中设置root用户密码（建议使用强密码）
# 3. 配置MySQL服务端口（默认3306）
# 4. 完成安装后，将MySQL的bin目录添加到系统PATH环境变量

验证安装

mysql --version
# 应该显示类似: mysql  Ver 8.0.xx for Win64 on x86_64 (MySQL Community Server - GPL)

macOS平台安装

使用Homebrew安装

brew install mysql
brew services start mysql

安全配置

mysql_secure_installation

Linux平台安装

以Ubuntu为例：

sudo apt update
sudo apt install mysql-server
sudo systemctl start mysql
sudo mysql_secure_installation

2. 创建Flask应用数据库

-- 登录MySQL
mysql -u root -p

-- 创建专用数据库
CREATE DATABASE flask_demo CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

-- 创建应用专用用户
CREATE USER 'flask_user'@'localhost' IDENTIFIED BY 'secure_password';

-- 授权
GRANT ALL PRIVILEGES ON flask_demo.* TO 'flask_user'@'localhost';

-- 刷新权限
FLUSH PRIVILEGES;

第二部分：Flask连接MySQL数据库

1. 安装必要依赖

pip install pymysql flask-sqlalchemy

2. 基础配置

在app.py中添加数据库配置：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)

# 配置数据库连接
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://flask_user:secure_password@localhost:3306/flask_demo'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_ECHO'] = True  # 开发时显示SQL语句

db = SQLAlchemy(app)

3. 定义第一个模型

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)
    created_at = db.Column(db.DateTime, server_default=db.func.now())
    
    def __repr__(self):
        return f'<User {self.username}>'

class Post(db.Model):
    __tablename__ = 'posts'
    
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(120), nullable=False)
    content = db.Column(db.Text, nullable=False)
    user_id = db.Column(db.Integer, db.ForeignKey('users.id'))
    
    author = db.relationship('User', backref=db.backref('posts', lazy=True))
    
    def __repr__(self):
        return f'<Post {self.title}>'

4. 初始化数据库

创建init_db.py文件：

from app import app, db

with app.app_context():
    db.create_all()
    print("数据库表已创建!")

运行：

python init_db.py

5. 基本CRUD操作

创建记录

@app.route('/add_user')
def add_user():
    new_user = User(username='张三', email='zhangsan@example.com')
    db.session.add(new_user)
    db.session.commit()
    return '用户添加成功!'

查询记录

@app.route('/users')
def list_users():
    users = User.query.all()  # 获取所有用户
    return render_template('users.html', users=users)

@app.route('/user/<username>')
def show_user(username):
    user = User.query.filter_by(username=username).first_or_404()
    return render_template('user.html', user=user)

更新记录

@app.route('/update_email/<username>/<new_email>')
def update_email(username, new_email):
    user = User.query.filter_by(username=username).first()
    if user:
        user.email = new_email
        db.session.commit()
        return '邮箱更新成功!'
    return '用户不存在!'

删除记录

@app.route('/delete_user/<username>')
def delete_user(username):
    user = User.query.filter_by(username=username).first()
    if user:
        db.session.delete(user)
        db.session.commit()
        return '用户删除成功!'
    return '用户不存在!'

6. 高级查询示例

# 分页查询
@app.route('/posts/<int:page>')
def show_posts(page=1):
    per_page = 10
    pagination = Post.query.order_by(Post.id.desc()).paginate(page, per_page, error_out=False)
    return render_template('posts.html', pagination=pagination)

# 复杂查询
@app.route('/search')
def search():
    keyword = request.args.get('q', '')
    results = Post.query.filter(
        (Post.title.contains(keyword)) | 
        (Post.content.contains(keyword))
    ).join(User).filter(User.username == '张三').all()
    return render_template('search.html', results=results)

7. 数据库迁移配置

安装Flask-Migrate：

pip install flask-migrate

配置迁移：

from flask_migrate import Migrate

migrate = Migrate(app, db)

初始化迁移仓库：

flask db init

创建迁移脚本：

flask db migrate -m "initial migration"

应用迁移：

flask db upgrade

第三部分：最佳实践与安全建议

连接池配置

app.config['SQLALCHEMY_POOL_SIZE'] = 20
app.config['SQLALCHEMY_POOL_TIMEOUT'] = 300
app.config['SQLALCHEMY_POOL_RECYCLE'] = 3600

敏感信息保护

使用环境变量存储数据库密码

创建.env文件：

DB_USER=flask_user
DB_PASS=secure_password

修改配置：

import os
from dotenv import load_dotenv

load_dotenv()
app.config['SQLALCHEMY_DATABASE_URI'] = f"mysql+pymysql://{os.getenv('DB_USER')}:{os.getenv('DB_PASS')}@localhost:3306/flask_demo"

性能优化

# 批量插入
db.session.bulk_insert_mappings(User, [
    {'username': 'user1', 'email': 'user1@example.com'},
    {'username': 'user2', 'email': 'user2@example.com'}
])

# 只查询需要的列
users = db.session.query(User.username, User.email).all()

错误处理

@app.route('/safe_delete/<int:user_id>')
def safe_delete(user_id):
    try:
        user = User.query.get_or_404(user_id)
        db.session.delete(user)
        db.session.commit()
        return '删除成功'
    except Exception as e:
        db.session.rollback()
        app.logger.error(f'删除用户失败: {str(e)}')
        return '删除失败', 500