mircealungu
mircealungu

Reputation: 6981

drop_all() freezes in Flask with SQLAlchemy

I am writing test cases for a Flask application.

I have a setUp method which drops the tables in the db before re-creating them again. It looks like this:

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

This works fine for the first test, but it freezes at drop_all before the second test is run.

EDIT: The stack trace looks like this when interrupting the process

  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)

Anybody has a clue how to fix this?

Upvotes: 42

Views: 19568

Answers (4)

Egor
Egor

Reputation: 359

Just close all sessions in your app and after that invoke 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)

more information about Sessions in SQLAlchemy http://docs.sqlalchemy.org/en/latest/orm/session_api.html?highlight=close_all

Upvotes: 33

Mauricio Souza Lima
Mauricio Souza Lima

Reputation: 608

I had the same problem, in my case I had 2 different sessions making queries to the same table. My solution was to use one scoped_session for both places.

I created it in a different module so I had no problem with circular dependencies, like this:

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)

Using only the db.session in all your code will guarantee that you are in the same session. In the tests, make sure you execute the rollback at tearDown.

Upvotes: 1

Brady Huang
Brady Huang

Reputation: 2002

I am a Flask developer and using flask_sqlalchemy and pytest to test my app server, I run into similar situation when I run the statement db.drop_all(), console shows that one of my table is locked.

I use db.session.remove()to remove the session before running db.drop_all().

Upvotes: 5

mircealungu
mircealungu

Reputation: 6981

Oki, there might be other solutions but for now, after searching the interwebs, I found that the problem disappears if I prepend my code with a myapp.db.session.commit(). I guess, somewhere a transaction was waiting to be committed.

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

Upvotes: 45

Related Questions