【问题标题】:SQLAlchemy object throwing attribute error for _sa_instance_state in pytestpytest中_sa_instance_state的SQLAlchemy对象抛出属性错误
【发布时间】:2021-01-05 20:34:20
【问题描述】:

我找不到任何其他讨论获取 SQLAlchemy 对象的“sa_instance_state”属性错误的线程。在 pytest 方法中调用 Voterlist.query.all() 时出现此错误。当我在 API 或其他任何地方调用 Voterlist.query.all() 时,我没有收到此错误。

这是一个带有前端和 api 蓝图的 Flask 应用程序。前端调用 API。

conftest.py:

class TestConfig(Config):                                                                                                                                   
    TESTING = True                                                                                                                                          
    SQLALCHEMY_DATABASE_URI = 'sqlite:///test.db'                                                                                                           
    WTF_CSRF_ENABLED = False                                                                                                                                
    BCRYPT_LOG_ROUNDS = 4                                                                                                                                   
                                                                                                                                                            
def fill_database():                                                                                                                                        
    s0 = Source(title='Catalist')                                                                                                                           
    s1 = Source(title='L2')                                                                                                                                 
    sources = [s0, s1]                                                                                                                                      
    l0 = Voterlist(description='Test List 0')                                                                                                               
    l1 = Voterlist(description='Test List 1')                                                                                                               
    lists = [l0, l1]                                                                                                                                        
    return lists, sources                                                                                                                                   
                                                                                                                                                            
@pytest.fixture(scope='class')                                                                                                                              
def test_client():                                                                                                                                          
    flask_app = create_app(TestConfig)                                                                                                                      
    with flask_app.test_client() as client:                                                                                                                 
        app_context = flask_app.test_request_context()                                                                                                      
        app_context.push()                                                                                                                                  
        db.create_all()                                                                                                                                     
        lists, sources = fill_database()                                                                                                                    
        for s in sources:                                                                                                                                   
            db.session.add(s)                                                                                                                               
            db.session.commit()                                                                                                                             
        for l in lists:                                                                                                                                     
            db.session.add(l)                                                                                                                        
            db.session.commit()                                                                                                                             
            sleep(1)                                                                                                                                        
        yield client                                                                                                                                        
        db.drop_all()                                                                                                                                       
    app_context.pop() 

test_api.py:

@pytest.mark.usefixtures('test_client')                                                                                                                     
class TestLists():                                                                                                                                          
    def test_get_lists(self, test_client):                                                                                                                  
        '''Test download all lists.'''                                                                                                                      
        response = test_client.get(url_for('api.lists'))                                                                                                    
        resplists = response.json['lists']                                                                                                                  
        assert type(resplists) == list                                                                                                                      
        dblists = Voterlist.query.all() <<<< Error here

API方法:

class ListAPI(Resource):                                                                                                                                    
    def __init__(self):                                                                                                                                     
        self.reqparse = reqparse.RequestParser()                                                                                                            
        self.fields = ['description', 'vendorquery', 'campaign', 's3key',                                                                                   
                       'ftimestamp', 'fname', 'source', 'user']                                                                                             
        for f in self.fields:                                                                                                                               
            self.reqparse.add_argument(f,                                                                                                                   
                                       type=str,                                                                                                            
                                       location='json')                                                                                                     
        super(ListAPI, self).__init__()                                                                                                                     
                                                                                                                                                            
    def get(self):                                                                                                                                          
        """Returns all lists."""                                                                                                                            
        try:                                                                                                                                                
            alllists = Voterlist.query.all()                                                                                                                
            lists = []                                                                                                                                      
            for l in alllists:                                                                                                                              
                l = l.__dict__                                                                                                                              
                l.pop('_sa_instance_state', None)                                                                                                           
                lists.append(l)                                                                                                                             
            return jsonify({'lists': lists})                                                                                                                
        except Exception as e:                                                                                                                              
            print(e)                                                                                                                                        
            abort(400, str(e)) 

您可以看到,当 API 方法调用 Voterlist.query.all() 时,它可以工作,甚至清楚地包含 _sa_instance_state 属性,否则当我从列表中的每个对象中弹出该属性时,它会抛出相同的错误。

这是 Voterlist 对象的定义:

class Voterlist(db.Model):
    id = db.Column(db.Integer, primary_key=True, default=get_time)
    description = db.Column(db.String(128))
    vendorquery = db.Column(db.String(128))
    campaign = db.Column(db.String(128))
    s3key = db.Column(db.String(128))
    fname = db.Column(db.String(128))
    ftimestamp = db.Column(db.Integer)
    source = db.Column(db.String(128))
    user = db.Column(db.String(128))
    status = db.Column(db.Integer, default=0)

    def __repr__(self):
        return '<List {}: {}>'.format(self.id, self.campaign)

以下是 pytest 的完整输出,以防万一:

=================================================================== test session starts ====================================================================
platform linux -- Python 3.6.9, pytest-6.0.1, py-1.9.0, pluggy-0.13.1
rootdir: /home/alazar66/Downloads/data/rl/rlabs-voter-list-mgr
plugins: anyio-2.0.2
collected 1 item                                                                                                                                           

tests/functional/test_api.py F                                                                                                                       [100%]

========================================================================= FAILURES =========================================================================
_________________________________________________________________ TestLists.test_get_lists _________________________________________________________________

self = <test_api.TestLists object at 0x74c48f8ff208>, test_client = <FlaskClient <Flask 'app'>>

    def test_get_lists(self, test_client):
        '''Test download all lists.'''
        response = test_client.get(url_for('api.lists'))
        resplists = response.json['lists']
        assert type(resplists) == list
>       dblists = Voterlist.query.all()

tests/functional/test_api.py:15: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/home/alazar66/.local/lib/python3.6/site-packages/sqlalchemy/orm/query.py:3373: in all
    return list(self)
/home/alazar66/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py:100: in instances
    cursor.close()
/home/alazar66/.local/lib/python3.6/site-packages/sqlalchemy/util/langhelpers.py:70: in __exit__
    with_traceback=exc_tb,
/home/alazar66/.local/lib/python3.6/site-packages/sqlalchemy/util/compat.py:182: in raise_
    raise exception
/home/alazar66/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py:80: in instances
    rows = [proc(row) for row in fetch]
/home/alazar66/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py:80: in <listcomp>
    rows = [proc(row) for row in fetch]
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

row = (1609879216, 'Test List 0', None, None, None, None, None, None, None, 0)

    def _instance(row):
    
        # determine the state that we'll be populating
        if refresh_identity_key:
            # fixed state that we're refreshing
            state = refresh_state
            instance = state.obj()
            dict_ = instance_dict(instance)
            isnew = state.runid != runid
            currentload = True
            loaded_instance = False
        else:
            # look at the row, see if that identity is in the
            # session, or we have to create a new one
            identitykey = (
                identity_class,
                tuple([row[column] for column in pk_cols]),
                identity_token,
            )
    
            instance = session_identity_map.get(identitykey)
    
            if instance is not None:
                # existing instance
>               state = instance_state(instance)
E               AttributeError: 'Voterlist' object has no attribute '_sa_instance_state'

/home/alazar66/.local/lib/python3.6/site-packages/sqlalchemy/orm/loading.py:532: AttributeError
================================================================= short test summary info ==================================================================
FAILED tests/functional/test_api.py::TestLists::test_get_lists - AttributeError: 'Voterlist' object has no attribute '_sa_instance_state'
==================================================================== 1 failed in 3.89s =====================================================================```

【问题讨论】:

    标签: python python-3.x flask sqlalchemy pytest


    【解决方案1】:

    我想通了。出于 API 调用方法的方便,我不恰当地从对象 __dict__ 属性中删除了“_sa_instance_state”,并且该删除在 API 调用之外仍然存在。

    我重写了 API 调用,并修复了它。

        def get(self):
            """Returns all lists."""
            try:
                alllists = Voterlist.query.all()
                lists = []
                for l in alllists:
                    l = {k: v for (k, v) in l.__dict__.items()
                         if k != '_sa_instance_state'}
                    lists.append(l)
                return jsonify({'lists': lists})
            except Exception as e:
                print(e)
                abort(400, str(e))
    

    【讨论】:

      猜你喜欢
      • 2016-01-10
      • 2018-12-23
      • 2021-12-27
      • 2022-11-06
      • 2020-07-08
      • 1970-01-01
      • 1970-01-01
      • 2013-04-09
      • 2013-04-15
      相关资源
      最近更新 更多