flask中使用 SQLAlchemy session的最佳实践

flask中使用SQLAlchemy最方便的方式是使用Flask-SQLAlchemy扩展，这个扩展实现了scoped session，据说就是在接收到请求时创建session，在处理完请求返回结果时close session，这样在请求的处理函数中可以直接使用session而不关心其创建、关闭、甚至是发生异常时的关闭过程。有其方便之处，但也带来了不方便的地方。其中笔者比较厌烦的有几点：

1. 数据库模型类变了
2. 每个请求都会创建session

这是 SQLAlchemy 的模型类和 Flask-SQLAlchemy 的模型类

# SQLAlchemy 模型类

from sqlalchemy import Column, DateTime, Float, Index, Integer
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()
metadata = Base.metadata

class UserCoin(Base):
    __tablename__ = 'user_coin'

    uid = Column(Integer, primary_key=True)
    coin_type = Column(String(8), nullable=False)
    balance = Column(Float, nullable=False, server_default=text("'0'"))
    create_time = Column(DateTime, nullable=False, server_default=text("CURRENT_TIMESTAMP"))
    address = Column(Text, nullable=False)

# Flask-SQLAlchemy 模型类

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////tmp/test.db'
db = SQLAlchemy(app)

class UserCoin(db.Model):
    __tablename__ = 'user_coin'

    uid = db.Column(db.Integer, primary_key=True)
    coin_type = db.Column(db.String(8), nullable=False)
    balance = db.Column(db.Float, nullable=False, server_default=text("'0'"))
    create_time = db.Column(db.DateTime, nullable=False, server_default=text("CURRENT_TIMESTAMP"))
    address = db.Column(db.Text, nullable=False)

可以看到基类变了Column变了，Integer String Float等类型都是属于db（flask_sqlalchemy.SQLAlchemy对象），这会有什么影响呢，笔者有时会使用sqlacodegen工具根据数据库里的schema反向生成SQLAlchemy模型，如果使用Flask-SQLAlchemy就需要手动转换一下了。

除非是管理后台之类的程序才会是每个请求都访问数据库。大部分应用是有一些请求访问数据库一些访问缓存，有可能就是request一来就创建session处理时又不需要之后再释放session，这白白浪费了资源，程序会执行无意义的操作，这种浪费随着访问数据库的request在总的request中的比例增长而线性增长。在做手机游戏时此比例达到巅峰，比如棋牌类的游戏（斗地主，牛牛，德州扑克等），大量的操作都是访问缓存中的数据，只是在游戏结束时才保存结果到数据库。

实践

如果在request处理函数中创建session，使用后再释放，代码如下：

@app.route('/<tournament_id>', methods=['POST'])
@general.usession
def create_tournament(tournament_id):
    """创建比赛"""
    session = Session()
    
    try:
        if args.uid != "abcd12345":
            return
            
        # logic code
        
        sess.commit()
    except Exception:
        sess.rollback()
    finally:
        sess.close()

每个处理函数都得写成这个样子，业务代码都得写到try中，使用with也是类似。最好是写在某段公用的代码中在某个时机自动执行释放。可以在处理函数之上使用装饰器声明“该请求的处理方法需要使用session”，在处理函数中就可以通过某个方法获取到session并使用，代码如下：

# general.py
from flask import g

# 装饰器 声明使用session
def usession(f):
    def func(*args, **kwargs):
        g.sess = Session()
        try:
            return f(*args, **kwargs)
        finally:
            g.sess.close()
    return func

# 获取session
def getsession():
    return g.sess

# main.py
from flask import Flask(__name__)
import general
app = Flask(__name__)

@app.route('/<tournament_id>', methods=['POST'])
@general.usession
def create_tournament(tournament_id):
    """创建比赛"""
    sess = general.getsession()
    
    tour = models.Tournament()
    tour.id = tournament_id
    tour.name = "test"
    
    sess.add(tour)
    sess.commit()
    return ""

如果不需要session，就不用usession装饰处理函数。