Drop_all() зависает в Flask с SQLAlchemy

Я пишу тестовые примеры для приложения Flask.

У меня есть метод setUp, который отбрасывает таблицы в db перед повторным созданием их. Это выглядит так:

def setUp(self):
    # other stuff...
    myapp.db.drop_all()
    myapp.db.create_all()
    # db creation...

Это отлично работает для первого теста, но задерживается на drop_all до запуска второго теста.

EDIT: Трассировка стека выглядит так, когда прерывается процесс

  File "populate.py", line 70, in create_test_db
    print (myapp.db.drop_all())
  File ".../flask_sqlalchemy/__init__.py", line 864, in drop_all
    self._execute_for_all_tables(app, bind, 'drop_all')
  File ".../flask_sqlalchemy/__init__.py", line 848, in _execute_for_all_tables
    op(bind=self.get_engine(app, bind), tables=tables)
  File ".../sqlalchemy/sql/schema.py", line 3335, in drop_all
  ....
  File "/Library/Python/2.7/site-packages/MySQLdb/cursors.py", line 190, in execute
    r = self._query(query)

Кто-нибудь знает, как это исправить?

Ответ 1

Oki, могут быть и другие решения, но пока, после поиска в interwebs, я обнаружил, что проблема исчезает, если я добавлю свой код с помощью myapp.db.session.commit(). Я думаю, где-то транзакция ожидала, что она будет совершена.

def setUp(self):
    # other stuff...
    myapp.db.session.commit()   #<--- solution!
    myapp.db.drop_all()
    myapp.db.create_all()
    # db creation...

Ответ 2

Просто закройте все сессии в вашем приложении и после вызова drop_all

    def __init__(self, conn_str):
        self.engine = create_engine(conn_str)
        self.session_factory = sessionmaker(engine)

    def drop_all(self):
        self.session_factory.close_all() # <- don't forget to close
        Base.metadata.drop_all(self._engine)

дополнительная информация о сеансах в SQLAlchemy http://docs.sqlalchemy.org/en/latest/orm/session_api.html?highlight=close_all

Ответ 3

Я являюсь разработчиком Flask и использую flask_sqlalchemy и pytest для тестирования моего сервера приложений. Я сталкиваюсь с подобной ситуацией, когда запускаю инструкцию db.drop_all(), консоль показывает, что одна из моих таблиц заблокирована.

Я использую db.session.remove() чтобы удалить сеанс перед запуском db.drop_all().

Ответ 4

У меня была та же проблема, в моем случае у меня было 2 разных сеанса, делающих запросы к одной и той же таблице. Мое решение состояло в том, чтобы использовать один scoped_session для обоих мест.

Я создал его в другом модуле, поэтому у меня не было проблем с круговыми зависимостями, например:

db.py:

from flask.ext.sqlalchemy import SQLAlchemy
db = SQLAlchemy()

models.py:

from .db import db
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True)

app.py:

from flask import Flask
from .db import db
app = Flask(__name__)
db.init_app(app)

Использование только db.session во всем вашем коде гарантирует, что вы находитесь в одном сеансе. В тестах убедитесь, что вы выполняете откат при tearDown.