入门Flask(四)数据库

安装

MySQL官网下载dmg安装

Homebrew安装(brew install mysql)
常用命令：

查看MySQL服务状态

# Homebrew安装的MySQL（Mac）
brew services list
# Linux系统
systemctl status mysqld

启动MySQL服务

brew services start mysql
# Linux系统
systemctl start mysqld   # 启动

停止MySQL服务

brew services stop mysql  # 权限不足加 sudo：sudo brew services stop mysql
# 重启 MySQL 服务（推荐，配置修改后用）
brew services restart mysql  # 权限不足加 sudo：sudo brew services restart mysql
# Linux系统
systemctl stop mysqld    # 停止
systemctl restart mysqld # 重启

本地登录（默认端口 3306）

mysql -u 用户名 -p
# 示例：root用户登录
mysql -u root -p

使用 SQLAlchemy 操作数据库

安装

使用 Flask-SQLAlchemy 连接 MySQL，需要两个核心依赖：Flask-SQLAlchemy（Flask 数据库 ORM 核心）、MySQL Python 驱动;
SQLite无需安装额外驱动（Python 内置 sqlite3，Flask-SQLAlchemy 直接兼容）
```
uv add flask-sqlalchemy
# 使用MySQL Python 驱动
uv add pymysql
```

初始化

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import DeclarativeBase

app = Flask(__name__)
class Base(DeclarativeBase):
  pass
db = SQLAlchemy(app, model_class=Base)

设置数据库URI

SQLALCHEMY_DATABASE_URI 遵循数据库类型 + 驱动:// 用户名：密码 @主机：端口 / 数据库名？连接参数的固定格式
sqlite:///后接相对路径 / 绝对路径 + 文件名，3 个斜杠是相对路径，4 个斜杠是绝对路径（sqlite:////）
直接指定数据库文件名，自动在项目根目录创建

from pathlib import Path
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////' + str(Path(app.root_path) / 'data.db')

MySQL:
必须手动创建 MySQL 数据库，SQLAlchemy 仅能自动创建表，无法创建库
登录MySQL后执行，创建指定数据库
CREATE DATABASE IF NOT EXISTS flask_mysql_db DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

app.config['SQLALCHEMY_DATABASE_URI'] = "mysql+pymysql://<用户名>:<密码>@<主机IP>:<端口>/<数据库名>?charset=utf8mb4&sql_mode=strict_trans_tables"

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = "mysql+pymysql://root:@localhost:3306/flask_mysql_db?charset=utf8mb4&sql_mode=strict_trans_tables"
# 关闭对象修改追踪功能
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

创建数据库模型

class User(db.Model):
    __tablename__ = 'user' # 定义表名称
    id: Mapped[int] = mapped_column(primary_key=True)  # 主键
    name: Mapped[str] = mapped_column(String(20))  # 名字

class Movie(db.Model):  # 表名将会是 movie
    __tablename__ = 'movie'
    id: Mapped[int] = mapped_column(primary_key=True)  # 主键
    title: Mapped[str] = mapped_column(String(60))  # 电影标题
    year: Mapped[str] = mapped_column(String(4))  # 电影年份

@app.route('/')
def index():
    user = db.session.execute(select(User)).scalar()  # 读取用户记录
    movies = db.session.execute(select(Movie)).scalars().all()  # 读取所有电影记录
    return render_template('index.html', user=user, movies=movies)

上面是SQLAlchemy 2.x 之后的声明式写法，下面是1.x的写法，除了基类创建变了，写法是兼容的

class User(db.Model):
    __tablename__ = 'user' # 定义表名称
    id = db.Column(db.Integer, primary_key=True)  # 主键
    name = db.Column(db.String(20))  # 名字

创建数据表

@app.cli.command('init-db')  # 注册为命令，传入自定义命令名
@click.option('--drop', is_flag=True, help='Create after drop.')  # 设置选项
def init_database(drop):
    """Initialize the database."""
    if drop:  # 判断是否输入了选项
        db.drop_all()
    db.create_all()
    click.echo('Initialized database.')  # 输出提示信息

生成虚拟数据

@app.cli.command()
def forge():
    """Generate fake data."""
    db.drop_all()
    db.create_all()

    # 全局的两个变量移动到这个函数内
    name = 'Grey Li'
    movies = [
        {'title': 'My Neighbor Totoro', 'year': '1988'},
        {'title': 'Dead Poets Society', 'year': '1989'},
        {'title': 'A Perfect World', 'year': '1993'},
    ]

    user = User(name=name)
    db.session.add(user)
    for m in movies:
        movie = Movie(title=m['title'], year=m['year'])
        db.session.add(movie)

    db.session.commit()
    click.echo('Done.')

参考

Flask入门教程