• 使用 SQLAlchemy：定義模型，配置數據庫，執行基本的 CRUD 操作。
• 創建和管理數據庫：使用 db.create_all() 創建表。
• CRUD 操作：添加、讀取、更新和刪除記錄。
• 查詢操作：執行基本和復雜查詢，包括排序和分頁。
• Flask-Migrate：使用 Flask-Migrate 管理數據庫遷移。
• 執行原始 SQL：使用原始 SQL 語句進行數據庫操作

一、準備工作

pip install flask flask_sqlalchemy

二、最小化可運行示例

from flask import Flask
from flask_sqlalchemy import SQLAlchemyapp = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///mydb.sqlite3'
#或者
#app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://user:password@localhost/db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = Falsedb = SQLAlchemy(app)# 定義數據模型（表）
class User(db.Model):id = db.Column(db.Integer, primary_key=True)username = db.Column(db.String(80), unique=True, nullable=False)age = db.Column(db.Integer)# 創建表
with app.app_context():db.create_all()# 啟動應用
@app.route('/')
def index():return '數據庫已初始化'

部分	含義
'sqlite:///mydatabase.db'	使用 SQLite 數據庫，數據庫文件保存在當前目錄
SQLAlchemy(app)	綁定 Flask 應用，初始化 ORM 引擎。也就是將 Flask 應用對象 app 注冊到 SQLAlchemy 對象中，從而啟用數據庫功能。
SQLALCHEMY_TRACK_MODIFICATIONS	關閉事件監聽，提升性能，防止警告

? 補充

延遲綁定方式（推薦方式）

適用于大型項目或避免循環引用：

# extensions.py
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy() # 先不綁定 app# app.py
from flask import Flask
from extensions import dbdef create_app():app = Flask(__name__)app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///mydatabase.db'db.init_app(app) # ? 后綁定 appreturn app

寫法	含義
db = SQLAlchemy(app)	初始化并立刻綁定 app
db = SQLAlchemy() + db.init_app(app)	更靈活，推薦在大型項目或工廠模式中使用

三、數據庫基本操作（增刪改查）

1. 插入數據（增）

with app.app_context():user = User(username='alice', age=20)db.session.add(user)db.session.commit()

db.session.add(user)：將新用戶對象添加到會話中。
db.session.commit()：提交事務，將更改保存到數據庫。

2. 查詢數據（查）

# 查詢所有
User.query.all()# 查詢第一條
User.query.first()# 條件查詢
User.query.filter_by(username='alice').first()# 更復雜的篩選
User.query.filter(User.age >= 18).all()

User.query.all()：查詢所有用戶記錄。

User.query.all()時報錯，RuntimeError: Working outside of application context.
說明你試圖在沒有 Flask 應用上下文的環境中訪問數據庫。這是 Flask-SQLAlchemy 的一個常規限制。

正確解決方案：
示例1、直接加上 app.app_context()

import Userwith app.app_context():users = User.query.all()print(users) 

示例2、在 Python 交互模式中（REPL 或 Jupyter Notebook）

from your_project import app from your_project.models import
Userapp.app_context().push()  # 手動推入應用上下文 User.query.all()

示例 3：在 Flask 視圖函數中（上下文自動存在）

 @app.route("/users") def get_users():users = User.query.all()return jsonify([u.to_dict() for u in users]) 

🔍 為什么必須有應用上下文？
Flask-SQLAlchemy 底層依賴 current_app 和 g，這些變量必須在上下文中才存在

📌如果你使用的是“應用工廠模式”（create_app()），記得這樣寫：

app = create_app() with app.app_context():db.create_all()User.query.all() 

3. 更新數據（改）

with app.app_context():user = User.query.filter_by(username='alice').first()if user:user.age = 25db.session.commit()

@app.route('/update_user/<int:user_id>')
def update_user(user_id):user = User.query.get(user_id)if user:user.username = 'new_username'db.session.commit()return 'User updated!'return 'User not found!'

User.query.get(user_id)：通過主鍵查詢單個用戶記錄。
更新字段值并提交事務。

4. 刪除數據（刪）

with app.app_context():user = User.query.filter_by(username='alice').first()if user:db.session.delete(user)db.session.commit()

db.session.delete(user)：刪除用戶記錄，并提交事務。

四、其他有用方法

方法 / 類	功能
db.Model	所有模型基類
db.Column	定義字段類型
db.session.add(obj)	添加記錄
db.session.commit()	提交更改
User.query.filter(...)	ORM 查詢
db.create_all()	創建所有表
db.drop_all()	刪除所有表

五、常用字段類型

類型	含義
db.String(n)	字符串（最長 n）
db.Integer	整數
db.Boolean	布爾值
db.DateTime	時間戳
db.Float	浮點數

六、初始化數據庫腳本（推薦）

可以加一個 Python 腳本 init_db.py，自動初始化數據庫：

from app import db
from app import app  # 確保 Flask app 存在with app.app_context():db.create_all()print("數據庫已初始化！")

db.create_all()：創建所有在當前上下文中定義的模型對應的表。

sqlalchemy 實例

基本使用

假設你有模型：

class User(db.Model):__tablename__ = 'users'id = db.Column(db.Integer, primary_key=True)username = db.Column(db.String(80), unique=True)

? __tablename__ = "users"在 SQLAlchemy 中,這是在 ORM 模型類中定義 對應的數據庫表名。
? 不加 __tablename__ 會怎樣？
SQLAlchemy 默認使用類名的小寫作為表名，如 User → user
但不推薦依賴默認行為 —— 明確指定更安全、更清晰
模型是數據庫表的 Python 類，每個模型類代表數據庫中的一張表。
? 推薦命名風格

模型類名	表名	建議寫法
User	users	__tablename__ = "users"
Order	orders	__tablename__ = "orders"
UserOrder	user_orders	__tablename__ = "user_orders"

你可以：

# 查詢所有用戶
users = db.session.query(User).all()# 查詢用戶名為 'root' 的用戶
user = db.session.query(User).filter_by(username='root').first()# 查詢 ID 大于 10 的用戶
users = db.session.query(User).filter(User.id > 10).all()

常見方法速查

方法	說明
query(Model)	創建查詢對象
.filter(condition)	過濾條件，支持表達式
.filter_by(field=value)	簡寫形式，等同于 == 比較
.first()	獲取第一條記錄或 None
.all()	獲取所有結果（列表）
.count()	獲取記錄數量
.limit(n)	限制返回數量
.offset(n)	跳過前 n 行記錄
.order_by(Model.field)	排序（可加 .desc()）
.one() / .one_or_none()	獲取一條記錄（不唯一會報錯）
.distinct()	去重

多表查詢（JOIN）

db.session.query(User, Post).join(Post, User.id == Post.user_id).all()

from sqlalchemy import or_users = User.query.filter(or_(User.username == 'john_doe', User.email == 'john@example.com')).all()

or_()：用于執行復雜的查詢條件。

原始 SQL 語句（可選）

from sqlalchemy import text
db.session.execute(text("SELECT * FROM users WHERE id = :id"), {"id": 1}).fetchall()

db.session.execute()：執行原始 SQL 查詢。

示例：分頁 + 排序

users = db.session.query(User) \.filter(User.username.like('%admin%')) \.order_by(User.id.desc()) \.offset(0).limit(10) \.all()

order_by()：按指定字段排序。
paginate()：分頁查詢。

推薦：使用 Flask-SQLAlchemy 提供的簡寫風格

User.query.filter_by(username='root').first()
User.query.filter(User.id > 10).all()

完整的 Flask-SQLAlchemy 配置示例（含連接池/超時/調試配置）

推薦項目結構（簡化）

your_project/
├── app.py
├── config.py           ← 配置集中管理
├── models.py           ← 數據模型
└── init_db.py          ← 初始化數據庫

? 1. config.py 配置文件

import osclass Config:# ? 數據庫連接（選你要的數據庫）# SQLite（開發測試用）SQLALCHEMY_DATABASE_URI = 'sqlite:///mydb.sqlite3'# MySQL 示例# SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://user:password@localhost:3306/mydb?charset=utf8mb4'# PostgreSQL 示例# SQLALCHEMY_DATABASE_URI = 'postgresql+psycopg2://user:password@localhost/mydb'# ? SQLAlchemy 設置SQLALCHEMY_TRACK_MODIFICATIONS = False# ? 調試設置DEBUG = True# ? 數據庫連接池設置SQLALCHEMY_ENGINE_OPTIONS = {"pool_size": 10,            # 最大連接數"max_overflow": 5,          # 超出 pool_size 后允許的連接數"pool_timeout": 30,         # 獲取連接最大等待時間（秒）"pool_recycle": 1800,       # 自動重連時間（秒）防止 MySQL 8 小時斷開"echo": True,               # 輸出執行 SQL（調試用）}# ? 密鑰（如啟用 Session）SECRET_KEY = os.environ.get('SECRET_KEY', 'dev-key')

數據庫連接池說明（MySQL/PostgreSQL 推薦）

參數	含義
pool_size	最大連接數
max_overflow	超出最大連接數后的額外連接數
pool_timeout	獲取連接等待時長（秒）
pool_recycle	多久重置連接（避免斷線），MySQL 默認 8 小時（28800 秒）斷連接，推薦 pool_recycle = 1800。
echo	是否打印 SQL（調試用）

? 2. app.py 應用初始化

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from config import Configdb = SQLAlchemy()def create_app():app = Flask(__name__)app.config.from_object(Config)db.init_app(app)# 注冊路由/藍圖@app.route('/')def index():return 'Hello Flask + SQLAlchemy'return app

? 3. models.py 數據模型定義

from app import dbclass User(db.Model):__tablename__ = 'users'id = db.Column(db.Integer, primary_key=True)username = db.Column(db.String(64), unique=True, nullable=False)age = db.Column(db.Integer)

? 4. init_db.py 數據庫初始化腳本

from app import create_app, db
from models import *app = create_app()with app.app_context():db.create_all()print("? 數據庫已初始化")

運行方式：

python init_db.py

使用 PyMySQL 連接 MySQL，app.py 文件代碼：

from flask import Flask, request, jsonify
import pymysqlapp = Flask(__name__)def get_db_connection():connection = pymysql.connect(host='localhost',user='username',password='password',database='dbname',cursorclass=pymysql.cursors.DictCursor)return connection@app.route('/add_user', methods=['POST'])
def add_user():data = request.jsonname = data['name']email = data['email']connection = get_db_connection()with connection.cursor() as cursor:cursor.execute('INSERT INTO user (username, email) VALUES (%s, %s)', (name, email))connection.commit()connection.close()return 'User added!'@app.route('/get_users')
def get_users():connection = get_db_connection()with connection.cursor() as cursor:cursor.execute('SELECT * FROM user')users = cursor.fetchall()connection.close()return jsonify(users)if __name__ == '__main__':app.run(debug=True)