【问题标题】:How do you resolve 'Already defined in this MetaData Instance' Error with Flask Pytest, SqlAlchemy, and Flask Sessions?如何使用 Flask Pytest、SqlAlchemy 和 Flask 会话解决“已在此元数据实例中定义”错误?
【发布时间】:2021-02-19 14:14:08
【问题描述】:

我在使用 SQLAlchemy、Flask(特别是使用 pytest 进行测试)和 Flask Sessions 库时遇到问题。

当我尝试设置基本测试客户端时,我收到以下警告 + 错误,导致所有测试无法正常工作:

SAWarning: This declarative base already contains a class with the same class name and module name as flask_session.sessions.Session, and will be replaced in the string-lookup table.
sqlalchemy.exc.InvalidRequestError: Table 'sessions' is already defined for this MetaData instance.  Specify 'extend_existing=True' to redefine options and columns on an existing Table object.

根据我所做的研究,如果我自己制作了这个会话模型,我会尝试向它添加extend_existing,这可能会解决问题,但它不在我创建的任何模型中,所以我不确定如何解决。

任何建议将不胜感激!谢谢。

init.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from flask import Flask
from flask import session
from flask_session import Session
from flask_sqlalchemy import SQLAlchemy

# Initialize database object
db=SQLAlchemy()
sess = Session()

def create_app(test=False):
    """Construct the core application."""
    app = Flask(__name__)
    if test:
        app.config.from_object('config.TestConfig')
    else:
        app.config.from_object('config.Config')
    
    # Configure db connection, initialize sessions
    db.app = app
    db.init_app(app)
    app.config['SESSION_SQLALCHEMY'] = db

    sess.init_app(app)
    with app.app_context():
        from . import routes  # Import routes
        db.create_all()  # Create database tables for our data models

        return app

test_basic.py

import os, sys
# sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# import main
import os
import tempfile
import flask
import pytest
from err_app import create_app,create_app_test, db, sess

import err_app
from flask_session import Session
from flask_sqlalchemy import SQLAlchemy

@pytest.fixture(scope='module')
def test_client():
    # flask_app = create_app_test()
    flask_app = create_app(test=True)
 
    # Establish an application context before running the tests.
    ctx = flask_app.app_context()
    ctx.push()
    if flask_app.config['TESTING'] == True:
        print("EMPTYING AND RE CREATING DATABASE")
        db.drop_all()
        from err_app.models import ErrorModel, ErrorParameterModel, TemplatesModel, UsersModel
        db.create_all()
        # sess.init_app(flask_app)
    # flask_app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///' + os.path.join(basedir, 'test.db')
 
    # Flask provides a way to test your application by exposing the Werkzeug test Client
    # and handling the context locals for you.
    testing_client = flask_app.test_client()

 
    yield testing_client  # this is where the testing happens!
 
    ctx.pop()

def test_create_app():
  app = create_app(test=True)
  assert app
  assert app.config['TESTING'] == True

def test_home_page(test_client):
    
    """
    GIVEN a Flask application
    WHEN the '/' page is requested (GET)
    THEN check the response is valid
    """
    
    response = test_client.get('/calculate_bounds')
    print(response)
    # print(response.text)
    assert response.status_code == 200

模型

from . import db
# from flask_sqlalchemy import SQLAlchemy
# db=SQLAlchemy()

class ErrorModel(db.Model):
    """Data model for ErrorModel accounts."""

    __tablename__ = 'err-models-table'
    id = db.Column(db.Integer,
                   primary_key=True)

    model_name = db.Column(db.Text,
                    index=False,
                    unique=False,
                    nullable=False)

    model_zip_attachment = db.Column(db.LargeBinary,
                                      index=False,
                                      unique=False,
                                      nullable=False)
    model_csv = db.Column(db.LargeBinary,
                            index=False,
                            unique = False,
                            nullable = False)
    created_at = db.Column(db.DateTime,
                            index=False,
                            unique=False,
                            nullable=False)

    model_version= db.Column(db.Text,
                              index=False,
                              unique=False,
                              nullable=False)

    app_version= db.Column(db.Text,
                            index=False,
                            unique=False,
                            nullable=False)

    notes = db.Column(db.Text,
                        index=False,
                        unique=False,
                        nullable=True)

    def __repr__(self):
        return '<ErrorModel {}>'.format(self.model_name)

class ErrorParameterModel(db.Model):
    """Data model for ErrorModel accounts."""

    __tablename__ = 'err-parameters'
    id = db.Column(db.Integer,
                   primary_key=True)

    error_description = db.Column(db.Text,
                    index=False,
                    unique=False,
                    nullable=False)

    error_label = db.Column(db.Text,
                    index=False,
                    unique=False,
                    nullable=False)
    template_id = db.Column(db.Integer,db.ForeignKey("templates.id", ondelete="CASCADE"))
    
    block_description = db.Column(db.Text,
                    index=False,
                    unique=False,
                    nullable=False)
    block_labels = db.Column(db.ARRAY(db.Text),
                    index=False,
                    unique=False,
                    nullable=False)
    input_type = db.Column(db.Text,
                    index=False,
                    unique=False,
                    nullable=False)
    uncorr_determ = db.Column(db.Text,
                    index=False,
                    unique=False,
                    nullable=False)

    value = db.Column(db.Text,
                    index=False,
                    unique=False,
                    nullable=False)

    unit = db.Column(db.Text,
                    index=False,
                    unique=False,
                    nullable=True)
    created_at = db.Column(db.DateTime,
                            index=False,
                            unique=False,
                            nullable=False)

    notes = db.Column(db.Text,
                        index=False,
                        unique=False,
                        nullable=True)

    def __repr__(self):
        return '<ErrorModel {}>'.format(self.error_description)

class TemplatesModel(db.Model):
    """Data model for ErrorModel accounts."""

    __tablename__ = 'templates'
    id = db.Column(db.Integer,
                   primary_key=True)

    name = db.Column(db.Text,
                    index=False,
                    unique=False,
                    nullable=False)

    user_id= db.Column(db.Integer,db.ForeignKey("users.id"))
    
    created_at = db.Column(db.DateTime,
                            index=False,
                            unique=False,
                            nullable=False)
    project = db.Column(db.Text,
                        index=False,
                        unique=False,
                        nullable=True)
    gauge_hw = db.Column(db.Text,
                        index=False,
                        unique=False,
                        nullable=True)

    error_model_version = db.Column(db.Text,
                        index=False,
                        unique=False,
                        nullable=True)
    model_number = db.Column(db.Text,
                        index=False,
                        unique=False,
                        nullable=True)
    num_sigmas = db.Column(db.Text,
                        index=False,
                        unique=False,
                        nullable=True)
    
    notes = db.Column(db.Text,
                        index=False,
                        unique=False,
                        nullable=True)

    children = db.relationship('ErrorParameterModel', cascade='all,delete', backref='templates')

    def __repr__(self):
        return '<Table {}>'.format(self.name)

class UsersModel(db.Model):
    """Data model for ErrorModel accounts."""

    __tablename__ = 'users'
    id = db.Column(db.Integer,
                 primary_key=True)

    created_at = db.Column(db.DateTime,
                            index=False,
                            unique=False,
                            nullable=False)

【问题讨论】:

    标签: session flask sqlalchemy pytest


    【解决方案1】:

    这是 Flask-Session 中的一个缺陷,已在未合并的拉取请求中修复:

    https://github.com/fengsp/flask-session/blob/1c1f7903184673682bd1d75432c8f455b62393a4/flask_session/sessions.py

    修复的主要部分是if table not in self.db.metadata:这一行

    但是,由于拉取请求太旧,它与最新版本略有不同。但是,如果您将 SqlAlchemySessionInterface 子类化,则修复本身仍然有效。这是我自己的 Flask 项目中当前正在修复的示例:https://github.com/tassaron2/flask-template/blob/8278afbbd18fcb398115fb1c7888010fc980dbad/app/main/plugins.py

    【讨论】:

      猜你喜欢
      • 2018-06-08
      • 2020-05-27
      • 2018-02-08
      • 2016-02-24
      • 2015-12-01
      • 1970-01-01
      • 2020-04-30
      • 2012-10-30
      • 1970-01-01
      相关资源
      最近更新 更多