Question: How to add a load_database functionality when using sqalchemy

[RosanneZe]

We started using this library recently and managed to get our tests working. The problem is that they are really slow, so I wanted to try and use something like the load_database that is mentioned in the documentation. The problem is that I have no idea how to do that in our setup. I was wondering if someone knows how to do this, or could point me to an example that is more relevant for our setup.

It's a pyramid app and the dbsession fixture that now works is:

from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker

@pytest.fixture
def dbsession(postgresql):
    connection = f'postgresql+psycopg2://{postgresql.info.user}:@{postgresql.info.host}:{postgresql.info.port}/{postgresql.info.dbname}'
    engine = create_engine(connection)

    session = scoped_session(sessionmaker(bind=engine))

    Base.metadata.create_all(engine)

    yield session

    Base.metadata.drop_all(engine)

[fizyk]

@RosanneZe the load parameter and load_database sample might be of help.
I haven't done it with SQLAlchemy however and it might be a walkthrough for both of us.

So first:
If we want to pre-load database based on SQLAlchemy models, my first thought would be to create all models in the load_database function, but drop them in the obsession fixture. The main issue could be that same models will be used in two different sessions... no session will not be necessary for the load_database just engine.

And this would result in more-or less this:

def load_database(**kwargs):
    connection = f"postgresql+psycopg2://{kwargs['user']}:@{kwargs['host']}:{kwargs['port']}/{kwargs['dbname']}"
    engine = create_engine(connection)
    Base.metadata.create_all(engine)


# Createa a process fixture referencing the load_database
postgresql_proc = factories.postgresql_proc(
    load=[load_database],
)


@pytest.fixture
def dbsession(postgresql):
    connection = f'postgresql+psycopg2://{postgresql.info.user}:@{postgresql.info.host}:{postgresql.info.port}/{postgresql.info.dbname}'
    engine = create_engine(connection)

    session = scoped_session(sessionmaker(bind=engine))

    Base.metadata.create_all(engine)  # should not be needed but If I recall, should not hurt. Candidate to be dropped anyway

    yield session

    Base.metadata.drop_all(engine)

[RosanneZe]

@fizyk Thank you so much! My apologies for the slow reply, I was on holiday. I've been playing around with it, and I can indeed remove the Base.metadata.create_all, but it doesn't really give a speed up. I think we need to add some data in the load_database to speed the tests up, so we don't need to add it for every test, but I'm not sure how to do that.

I tried just adding a session = scoped_session(sessionmaker(bind=engine)) to the load_database and using that, but that just froze my tests. Adding a session.flush() and a transaction.commit() did not help.

[fizyk]

@RosanneZe would you be able to provide a sample repository? Or PR here with sample behaviour?
I could check in my spare time and later that could be used as an example for pytest-postgresql 🤔

[fizyk]

@RosanneZe please take a look at the PoC I made at my sqlalchemy based project:

fizyk/pyramid_fullauth#721

[RosanneZe]

@fizyk thank you! That looks promising, I'll try to get it to work next week and let you know how it works out.

[RosanneZe]

We got it to work! This is what it looks like without pyramid_basemodel:

def load_database(**kwargs):
    connection = f"postgresql+psycopg2://{kwargs['user']}:@{kwargs['host']}:{kwargs['port']}/{kwargs['dbname']}"
    engine = create_engine(connection)
    Base.metadata.create_all(engine)
    session = scoped_session(sessionmaker(bind=engine))
    # add things to session
    session.commit()

postgresql_proc = factories.postgresql_proc(load=[load_database])

postgresql = factories.postgresql('postgresql_proc') # still need to check if this is actually needed or not

@pytest.fixture
def dbsession(postgresql):
    connection = f'postgresql+psycopg2://{postgresql.info.user}:@{postgresql.info.host}:{postgresql.info.port}/{postgresql.info.dbname}'
    engine = create_engine(connection)

    session = scoped_session(sessionmaker(bind=engine))

    yield session
    # 'Base.metadata.drop_all(engine)' here specifically does not work. It is also not needed. If you leave out the session.close()
    # all the tests still run, but you get a warning/error at the end of the tests.
    session.close()