在使用 Flask 和 Peewee 操作数据库时，确保数据库连接在使用后正确关闭非常重要。Peewee 提供了几种方式来自动管理数据库连接的生命周期。以下是几种常见的方法：

1.使用flask钩子可以在请求上下文前后开启和关闭数据库连接，相关钩子函数的笔记可以查看我之前的笔记：https://sulao.cn/post/666.html

from flask import Flask, g
from peewee import MySQLDatabase, Model, CharField

app = Flask(__name__)

# 配置数据库连接池
db = MySQLDatabase('sulao', user='root', password='password', host='localhost', port=3306)

class User(Model):
    name = CharField()

    class Meta:
        database = db

@app.before_request
def before_request():
    # 在请求开始时打开数据库连接
    db.connect()

@app.after_request
def after_request(response):
    # 在请求结束后关闭数据库连接
    if not db.is_closed():
        db.close()
    return response

@app.route('/')
def index():
    # 操作数据库
    user = User.get(User.name == 'sulao')
    return f"User: {user.name}"

if __name__ == '__main__':
    app.run()

@app.before_request钩子会在每个请求开始时打开数据库连接，@app.after_request` 钩子会在每个请求结束后关闭数据库连接，使用 db.is_closed()检查连接是否已关闭，避免重复关闭。

2.使用Peewee的flask_peewee扩展

flask_peewee是一个 Flask 扩展，专门用于集成 Peewee 和 Flask。它可以自动管理数据库连接的生命周期。

这个库我们需要安装一下

pip install flask-peewee

然后使用代码如下：

from flask import Flask
from flask_peewee.db import Database
from peewee import Model, CharField

app = Flask(__name__)
app.config['DATABASE'] = {
    'name': 'sulao',
    'engine': 'peewee.MySQLDatabase',
    'user': 'root',
    'password': 'password',
    'host': 'localhost',
    'port': 3306,
}

# 初始化数据库
db = Database(app)

class User(Model):
    name = CharField()

    class Meta:
        database = db.database

@app.route('/')
def index():
    # 操作数据库
    user = User.get(User.name == 'sulao')
    return f"User: {user.name}"

if __name__ == '__main__':
    app.run()

flask_peewee会自动管理数据库连接的生命周期，无需手动打开或关闭连接。

3.使用上下文管理器

from flask import Flask
from peewee import MySQLDatabase, Model, CharField

app = Flask(__name__)

# 配置数据库
db = MySQLDatabase('sulao', user='root', password='password', host='localhost', port=3306)

class User(Model):
    name = CharField()

    class Meta:
        database = db

@app.route('/')
def index():
    # 使用上下文管理器管理连接
    with db.connection_context():
        user = User.get(User.name == 'sulao')
        return f"User: {user.name}"

if __name__ == '__main__':
    app.run()

db.connection_context()会创建一个上下文管理器，确保在with块内数据库连接是打开的，并在块结束时自动关闭连接，这种方式非常适合在单个请求中操作数据库的场景。

4.使用Flask 的g对象

Flask 的g对象可以用于存储数据库连接，并在请求结束时关闭连接。

from flask import Flask, g
from peewee import MySQLDatabase, Model, CharField

app = Flask(__name__)

# 配置数据库
db = MySQLDatabase('sulao', user='root', password='password', host='localhost', port=3306)

class User(Model):
    name = CharField()

    class Meta:
        database = db

@app.before_request
def before_request():
    g.db = db
    g.db.connect()

@app.teardown_request
def teardown_request(exception):
    if hasattr(g, 'db') and not g.db.is_closed():
        g.db.close()

@app.route('/')
def index():
    # 操作数据库
    user = User.get(User.name == 'sulao')
    return f"User: {user.name}"

if __name__ == '__main__':
    app.run()

这种也是使用了钩子函数作为请求上下文前后来挂载数据库打开对象到g对象上进行操作，这种方式适合需要手动管理连接的场景。

5.使用Peewee的atomic上下文管理器

如果你需要在事务中操作数据库，可以使用Peewee的atomic上下文管理器。它不仅会管理事务，还会自动管理连接。

from flask import Flask
from peewee import MySQLDatabase, Model, CharField
app = Flask(__name__)
# 配置数据库
db = MySQLDatabase('sulao', user='root', password='password', host='localhost', port=3306)
class User(Model):
    name = CharField()
    class Meta:
        database = db
@app.route('/')
def index():
    # 使用 atomic 上下文管理器
    with db.atomic():
        user = User.get(User.name == 'sulao')
        return f"User: {user.name}"
if __name__ == '__main__':
    app.run()

db.atomic()会创建一个事务上下文管理器，确保在with块内数据库连接是打开的，并在块结束时自动关闭连接，适合需要事务支持的场景。