Flask无法读取SQLite数据库内容的问题排查

问题场景 #

已有一个通过sqlite3创建的SQLite数据库，表结构如下：

CREATE TABLE IF NOT EXISTS qa (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    subject VARCHAR(500),
    question VARCHAR(500),
    correct_answer VARCHAR(500),
    choice_a VARCHAR(500),
    choice_b VARCHAR(500),
    choice_c VARCHAR(500),
    choice_d VARCHAR(500)
);

用原生sqlite3查询能看到表中有149481条数据：

connection = sqlite3.connect('questions_and_answers.db')
cursor = connection.cursor()
cursor.execute("SELECT COUNT(*) FROM qa;")
print(cursor.fetchall())
# 输出：[(149481,)]

但将数据库文件放在Flask代码同目录下，用Flask-SQLAlchemy连接后，查询返回0条数据，相关Flask代码如下：

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import func, inspect
import sqlite3

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///questions_and_answers.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_ECHO'] = True

db = SQLAlchemy(app)

class Question(db.Model):
    __tablename__ = 'qa'
    id = db.Column(db.Integer, primary_key=True)
    subject = db.Column(db.String(500))
    question = db.Column(db.String(500))
    correct_answer = db.Column(db.String(500))
    choice_a = db.Column(db.String(500))
    choice_b = db.Column(db.String(500))
    choice_c = db.Column(db.String(500))
    choice_d = db.Column(db.String(500))
    
    def __repr__(self):
        return f'<Question {self.id}>'

with app.app_context():
    db.reflect()
    db.create_all()
    results = Question.query.all()
    print(len(results)) # 输出：0

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

解决步骤 #

1. 修正数据库路径（最可能的原因） #

Flask运行时的工作目录可能不是代码文件所在目录，导致SQLAlchemy创建了新的空数据库文件，而非连接已有的那个。改用绝对路径指定数据库：

import os
basedir = os.path.abspath(os.path.dirname(__file__))
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + os.path.join(basedir, 'questions_and_answers.db')

2. 移除不必要的操作 #

因为数据库和表已经存在，db.reflect()和db.create_all()在这里是多余的，甚至可能引发意外：

• db.reflect()：自动加载数据库表结构，若模型已定义可能导致冲突
• db.create_all()：虽不会覆盖已有表，但路径错误时会创建新的空表
  直接删除这两行即可。

3. 匹配模型与原表的自增设置 #

原表的id是AUTOINCREMENT，显式指定autoincrement=True能确保模型与原表结构完全匹配，避免映射问题：

id = db.Column(db.Integer, primary_key=True, autoincrement=True)

4. 查看SQLAlchemy日志验证查询 #

开启SQLALCHEMY_ECHO=True后，控制台会输出所有执行的SQL语句，可通过日志确认：

• SQLAlchemy连接的数据库文件路径是否正确
• 查询的表名和字段是否与原表一致

修改后的完整代码示例 #

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
import os

app = Flask(__name__)
# 用绝对路径指定数据库
basedir = os.path.abspath(os.path.dirname(__file__))
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + os.path.join(basedir, 'questions_and_answers.db')
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
app.config['SQLALCHEMY_ECHO'] = True

db = SQLAlchemy(app)

class Question(db.Model):
    __tablename__ = 'qa'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    subject = db.Column(db.String(500))
    question = db.Column(db.String(500))
    correct_answer = db.Column(db.String(500))
    choice_a = db.Column(db.String(500))
    choice_b = db.Column(db.String(500))
    choice_c = db.Column(db.String(500))
    choice_d = db.Column(db.String(500))
    
    def __repr__(self):
        return f'<Question {self.id}>'

with app.app_context():
    # 直接查询，不需要reflect和create_all
    results = Question.query.all()
    print(len(results)) # 现在应该输出149481

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

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