объект pytest / sqlalchemy не связан с ошибкой сеанса - PullRequest
Новая
       9

объект pytest / sqlalchemy не связан с ошибкой сеанса

1 голос
/ 23 октября 2019

Я прочитал несколько постов, подобных моей проблеме, но не нашел ничего актуального. У меня есть настольное приложение, и я пытаюсь выполнить некоторые тесты с помощью pytest. Для этого я запускаю сервер MySQL в контейнере Docker и использую sqlalchemy для создания сеанса для добавления, обновления и т. Д. Объектов. Однако у меня есть эта ошибка, я не могу понять, как исправить sqlalchemy.orm.exc.DetachedInstanceError: Instance <Device at 0x7f4a38a01fd0> is not bound to a Session; attribute refresh operation cannot proceed

dspt.db .__ init __. Py 

import os
from contextlib import contextmanager
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.ext.declarative import declarative_base


Base = declarative_base()
user = os.environ.get('MYSQL_USER')
pwd = os.environ.get('MYSQL_PWD')
url = os.environ.get('MYSQL_URL', 'localhost')
db_url = f"mysql+pymysql://{user}:{pwd}@{url}/db?charset=utf8mb4&binary_prefix=true"
print(db_url)
engine = create_engine(db_url, encoding='utf-8')
Session = scoped_session(sessionmaker(bind=engine))


@contextmanager
def db_session():
    """ Creates a SQLAlchemy session"""
    session = Session()
    try:
        yield session
        # session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.expunge_all()
        session.close()

test.py 

import pytest
from dspt.db import db_session, Base
from dspt.db.models import Device

@pytest.fixture(scope='session')
def get_device():
    return [Device(), Device(), Device()]


@pytest.fixture(scope='session')
def seed(get_device):
    devices = get_device
    with db_session() as session:
        for d in devices:
            session.add(d)
        session.commit()

def test_get_device(seed, get_device):
    with db_session() as session:
        devices = get_device
        res = session.query(Device)\
                     .filter(Device.serial == devices[0].serial)\
                     .all()

        assert len(res) == 1
...