最近做Flask開發，使用Flask-SQLAlchemy拓展連接資料庫，但是在單測中遇到一個很難搞的bug，莫名其妙報錯，測試運行中報錯找不到xxx表：sqlite3.OperationalError: no such table: xxx，以及其他的各種各樣一連串的錯誤。

這個bug雖然很快找到了解決方案，但我花了更多的時間去挖掘它背後的原因。雖然產生這個bug的情況比較罕見，估計很少有人會遇上相同的情況，但這次debug經歷我覺得挺有意思，所以記錄下來供以後參考。

出於某些原因，我在項目中用到了兩個app，即Flask對象，以及兩個db，即SQLAlchemy對象。同時，為了使用我自定義的元類，直接使用的是declarative_base方法生成的Model（這是導致bug的根本原因之一）。另外，由於我的單測用的是內存資料庫，每個testcase運行前都要新建表，運行後都要刪除表，在這種情況下，當多個單測分別涉及到不同的app和db時，就會產生開頭說的bug。

1、Bug復現及解決方案

我提煉出了產生該bug的較簡短的代碼，如下。在命令行運行python -m unittest discover，下面的代碼會產生一連串的bug，而且錯誤提示無法直接定位到真正產生bug的代碼處，給debug造成了很大的難度。

解決這個bug的方法其實很簡單，只需要把視圖函數中的Flask-SQLAlchemy庫的查詢語句換成SQLAlchemy庫原生的查詢語句即可，即把A.query.all()換成db_a.session.query(A).all()。將代碼中所有的查詢語句都換成第二種方式後，單測便無bug了。

import unittest
from flask import Flask
from flask_sqlalchemy import SQLAlchemy, DefaultMeta, declarative_base, Model

MyModel = declarative_base(
cls=Model,
name=MyModel,
metaclass=DefaultMeta
)
class Config:
DEBUG = True
SQLALCHEMY_DATABASE_URI = sqlite://
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_ECHO = False

app_a = Flask(app_a)
app_a.config.from_object(Config)
db_a = SQLAlchemy(app_a, model_class=MyModel)

app_b = Flask(app_b)
app_b.config.from_object(Config)
db_b = SQLAlchemy(app_b, model_class=MyModel)

class A(db_a.Model):
__tablename__ = a
id = db_a.Column(db_a.Integer, primary_key=True, autoincrement=True)
content = db_a.Column(db_a.Text, nullable=False)

class B(db_b.Model):
__tablename__ = b
id = db_b.Column(db_b.Integer, primary_key=True, autoincrement=True)
content = db_b.Column(db_b.Text, nullable=False)

@app_a.route(/a)
def a():
a_s = A.query.all()
#a_s = db_a.session.query(A).all()
return str([a.content for a in a_s])

@app_b.route(/b)def b():
bs = B.query.all()
#bs = db_b.session.query(B).all()
return str([b.content for b in bs])

class TestA(unittest.TestCase):
app = app_a
db = db_a

def setUp(self):
self.app.testing = True
self.client = self.app.test_client()
self.app_context = self.app.app_context()
self.app_context.push()
self.db.create_all()

a_s = [A(content=a%d%i)for i in range(10)]
self.db.session.add_all(a_s)
self.db.session.commit()

def tearDown(self):
self.db.session.remove()
self.db.drop_all()
self.db.engine.dispose()
self.app_context.pop()

def test1(self):
rsp = self.client.get(/a)

def test2(self):
rsp = self.client.get(/a)

class TestB(unittest.TestCase):
app = app_b
db = db_b

def setUp(self):
self.app.testing = True
self.client = self.app.test_client()
self.app_context = self.app.app_context()
self.app_context.push()
self.db.create_all()

bs = [B(content=b%d%i)for i in range(10)]
self.db.session.add_all(bs)
self.db.session.commit()

def tearDown(self):
self.db.session.remove()
self.db.drop_all()
self.db.engine.dispose()
self.app_context.pop()

def test1(self):
rsp = self.client.get(/b)

def test2(self):
rsp = self.client.get(/b)

2、Bug產生的原因分析

這個bug的直接原因是資料庫的session錯亂。

db.session是一個scoped_session對象，這是一個線程局部變數，每次調用db.session()時會得到一個當前使用中的session對象，調用db.session.remove()方法會關閉舊的session，之後再次調用db.session()則會得到一個全新的session。直接調用db.session.query使用的是對應db當前使用的session。並且，正常情況下，調用Model.query使用的也應該是對應db當前使用的session，從Model.query.session可以獲得該對象。

在上面的列子中運行下面的代碼，從結果可知，使用Model.query方式的查詢語句使用的是同一個session，而db.session.query方式的查詢語句則使用的是不同session。模型A和B屬於不同的db，本應是使用不同的session的，但是在前一種方式中卻錯誤的使用了同一個session。

在第一個單測移除後，會執行db.drop_all()以清除舊的數據，而當調用了db.session.remove()後，舊的被拋棄的session依然被Model.query使用，而舊session中的表已經被清空了，所以用舊session查詢會引發一連串的錯誤。

print(
A.query.session == B.query.session,
db_a.session.query(A) == db_b.session.query(B)
)

# 結果：
# True False

進一步的，運行下面的代碼可以發現，A.query和B.query使用的session竟然都是db_b的session，這樣就導致了session的混用。所以使用db.session.query的方式才能正確的區分不同db使用的session，纔不會有bug。

db_a.session.remove()
db_b.session.remove()

s1 = A.query.session
s2 = B.query.session
s3 = db_a.session()
s4 = db_b.session()
print(s1==s2, s1==s3, s1 == s4, s2==s3, s2==s4, s3==s4)

# 結果：
# True False True False True False

3、進一步的分析

另一個關鍵的原因是，這裡用的Model是用declarative_base方法生成的MyModel，而兩個db使用的基礎Model都是這一個。可以看到，在Flask-SQLAlchemy的源碼中，在實例化SQLAlchemy對象（db）的過程中，會給Model添加一個新的屬性model.query，_QueryProperty對象會保存一個SQLAlchemy對象（db），並使用該對象的session完成query操作。

在上面的代碼中我實例化了兩個db，但是使用的都是同一個Model，這導致前一次的_QueryProperty對象被替換了，保存了第二次實例化時的db，即db_b。所以最終A.query和B.query使用的是同一個db_b的session。

如果調用兩次declarative_base，產生不同的基本Model，給不同的db使用，也不會出現該bug。

# flask_sqlalchemy/__init__.py
......

def make_declarative_base(self, model, metadata=None):
"""......"""
if not isinstance(model, DeclarativeMeta):
model = declarative_base(
cls=model,
name=Model,
metadata=metadata,
metaclass=DefaultMeta
)

# if user passed in a declarative base and a metaclass for some reason,
# make sure the base uses the metaclass
if metadata is not None and model.metadata is not metadata:
model.metadata = metadata

if not getattr(model, query_class, None):
model.query_class = self.Query

model.query = _QueryProperty(self)
return model

......

class _QueryProperty(object):
def __init__(self, sa):
self.sa = sa

def __get__(self, obj, type):
try:
mapper = orm.class_mapper(type)
if mapper:
return type.query_class(mapper, session=self.sa.session())
except UnmappedClassError:
return None
......

經過漫長的探索，找到了產生bug真正的原因，這次debug過程中讓我感受最深刻的是：

Explicit is better than implicit.

顯式優於隱式，真的是太對了！

在SQLAlchemy中session可謂是貫穿整體的重要概念，所有關於資料庫的操作都必須在session中完成。但是Flask-SQLAlchemy提供了一種簡便的查詢方式，在查詢的過程中讓我們感受不到session的存在，而這種隱式調用的方法使得我們難以預料使用的session，某些情況下會造成極大的困惑。

在編程中要盡量顯式的調用重要的操作，除非它不值一提，或者隱式的實現可以帶來巨大的便利。

推薦閱讀：

相關文章