python - 将 SQLAlchemy 会话与 HTTPServer 一起使用。线程错误

Question

我在使用 SQLAlchemy 会话在不同线程上生成 HTTP 服务器时遇到问题（我在另一个线程上生成 HTTP 服务器的原因是因为否则测试会永远挂起）。发生的事情是我在每个请求中都传递了数据库配置。但它不允许我在使用 SQLite 的测试服务器上创建会话（开发服务器是 PostgreSQL）。每次它进入视图时都会引发以下错误：

sqlite3.ProgrammingError: SQLite objects created in a thread can only be used in that same thread.The object was created in thread id 5784 and this is thread id 8560

这是我的测试用例：

class ViewsTestSuite(unittest.TestCase):


    def setUp(self):
        app = App(8000, testing=True)
        try:
            httpd_thread = threading.Thread(target=app.run)
            httpd_thread.daemon = True
            httpd_thread.start()
        except KeyboardInterrupt:
            pass

我的应用程序设置类：

class App:

    def __init__(self, port, testing=False):
        self.port = port
        self.testing = testing
        self.server = None
        self.config = None
        self.db = None

    def run(self):
        if self.testing:
            self.config = TestingConfig
        else:
            self.config = DevelopmentConfig
        self.init_db()
        RequestHandler.config = self.config
        self.server = HTTPServer(('localhost', self.port), RequestHandler)
        try:
            self.server.serve_forever()
        except KeyboardInterrupt:
            pass
        self.server.server_close()

    def stop(self):
        self.server.server_close()

    def init_db(self):
        self.db = Database(self.config.DATABASE_URL)
        self.db.create_session()
        self.db.create_database()

和我正在测试的视图：

def get_recipes(handler):
    json_structure = []
    db = get_db(handler.config.DATABASE_URL)
    db.create_session()
    recipes = db.session.query(Recipe, func.coalesce(func.avg(Rating.rating).label('average'), 0)).\
        outerjoin(Rating).group_by(Recipe.id).all()
    for recipe in recipes:
        json_structure.append({
            "id": recipe[0].id,
            "name": recipe[0].name,
            "prep_time": recipe[0].prep_time,
            "vegeterian": recipe[0].vegeterian,
            "rating": float(recipe[1])
        })

    return 200, json.dumps(json_structure)

任何帮助都会非常棒！非常感谢

标签： pythonpython-3.xsqlalchemy