在Google App Engine中设置全局数据库连接的规范方案咨询

首先说下你现有代码的几个问题：

• 代码存在明显bug：global_conn = pymysql.connect(...)后没给db_conn赋值，global_cursor = conn.cursor()里的conn也未定义，实际运行会报错；
• 全局单连接+单cursor的方式在GAE环境下完全不适用：GAE是多请求并发的，单个连接无法被多个线程安全共享，会导致查询混乱、连接超时；
• 没有连接重连逻辑，数据库端会主动断开长时间闲置的连接，后续请求会直接失败。

下面是几种适配GAE环境的推荐方案：

方案一：用Flask-SQLAlchemy（最推荐） #

Flask-SQLAlchemy内置了成熟的连接池管理，能自动处理连接的创建、复用、重连，完美适配GAE的多实例/并发请求场景。

示例代码：

from flask import Flask, jsonify
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
# 配置MySQL连接（pymysql作为驱动）
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://x:y@w/z?charset=utf8'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# 配置连接池参数，避免连接被数据库断开
app.config['SQLALCHEMY_POOL_SIZE'] = 5  # 连接池最大连接数
app.config['SQLALCHEMY_POOL_RECYCLE'] = 300  # 5分钟自动回收连接

db = SQLAlchemy(app)

# 对应你的mytable表结构（按需定义）
class MyTable(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    # 其他字段根据实际表结构添加

@app.route('/')
def hello():
    return jsonify(hello='world')

@app.route('/db')
def db_query():
    # ORM查询，自动管理连接
    records = MyTable.query.limit(50).all()
    # 转成易返回的字典格式
    result = [{'id': item.id} for item in records]
    return jsonify(result)

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

方案二：手动用连接池（适合不想用ORM的场景） #

如果坚持用原生pymysql，可以用DBUtils.PooledDB实现线程安全的连接池，管理连接的复用。

先安装依赖：pip install DBUtils

示例代码：

import pymysql
from dbutils.pooled_db import PooledDB
from flask import Flask, jsonify

# 全局初始化连接池，只会执行一次
db_pool = PooledDB(
    creator=pymysql,
    maxconnections=5,  # 连接池最大容量
    mincached=1,       # 初始化时创建的空闲连接数
    host='w',
    user='x',
    password='y',
    database='z',
    charset='utf8'
)

app = Flask(__name__)

@app.route('/')
def hello():
    return jsonify(hello='world')

@app.route('/db')
def db_query():
    # 从连接池取连接
    conn = db_pool.connection()
    try:
        # 每次请求用独立的cursor
        with conn.cursor() as cursor:
            cursor.execute("SELECT * FROM mytable LIMIT 50")
            res = cursor.fetchall()
            # 把结果转成字典列表
            result = [dict(zip([desc[0] for desc in cursor.description], row)) for row in res]
        return jsonify(result)
    finally:
        # 把连接放回池，不是真的关闭
        conn.close()

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

应急修正方案（不推荐高并发场景） #

如果一定要沿用你原来的思路，至少要修复bug、添加重连逻辑，并且不要复用全局cursor：

import pymysql
from flask import Flask, jsonify

db_conn = None

def get_db():
    global db_conn
    try:
        if db_conn:
            # 检查连接是否存活，失效则自动重连
            db_conn.ping(reconnect=True)
        else:
            db_conn = pymysql.connect(user="x", password="y", database="z", host="w", charset='utf8')
        return db_conn
    except pymysql.MySQLError:
        # 连接彻底失效，重新创建
        db_conn = pymysql.connect(user="x", password="y", database="z", host="w", charset='utf8')
        return db_conn

app = Flask(__name__)

@app.route('/')
def hello():
    return jsonify(hello='world')

@app.route('/db')
def db():
    conn = get_db()
    # 每次请求创建新cursor，避免并发冲突
    with conn.cursor() as cursor:
        cursor.execute("SELECT * FROM mytable LIMIT 50")
        res = cursor.fetchall()
        result = [dict(zip([desc[0] for desc in cursor.description], row)) for row in res]
    return jsonify(result)

注意：这个方案只适合低并发场景，多请求同时访问时会出现线程安全问题。

内容的提问来源于stack exchange，提问作者David542