Skip to content

Instantly share code, notes, and snippets.

@einSelbst

einSelbst/sqlalchemy_conftest.py

Forked from kissgyorgy/sqlalchemy_conftest.py
Created March 14, 2018 12:19
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save einSelbst/6dd2d17e670f416271c45c4f27610e6f to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save einSelbst/6dd2d17e670f416271c45c4f27610e6f to your computer and use it in GitHub Desktop.
Download ZIP

Revisions

  1. @kissgyorgy kissgyorgy renamed this gist May 25, 2015. 1 changed file with 2 additions and 2 deletions.
    4 changes: 2 additions & 2 deletions conftest.py → sqlalchemy_conftest.py
    View file
    Original file line number Diff line number Diff line change
    @@ -21,14 +21,14 @@ def dbsession(engine, tables):
    """Returns an sqlalchemy session, and after the test tears down everything properly."""
    connection = engine.connect()
    # begin the nested transaction
    transacton = connection.begin()
    transaction = connection.begin()
    # use the connection with the already started transaction
    session = Session(bind=connection)

    yield session

    session.close()
    # roll back the broader transaction
    transacton.rollback()
    transaction.rollback()
    # put back the connection to the connection pool
    connection.close()
  2. @kissgyorgy kissgyorgy revised this gist May 24, 2015. 1 changed file with 3 additions and 0 deletions.
    3 changes: 3 additions & 0 deletions conftest.py
    View file
    Original file line number Diff line number Diff line change
    @@ -3,16 +3,19 @@
    from myapp.models import BaseModel
    import pytest


    @pytest.fixture(scope='session')
    def engine():
    return create_engine('postgresql://localhost/test_database)


    @pytest.yield_fixture(scope='session')
    def tables(engine):
    BaseModel.metadata.create_all(engine)
    yield
    BaseModel.metadata.drop_all(engine)


    @pytest.yield_fixture
    def dbsession(engine, tables):
    """Returns an sqlalchemy session, and after the test tears down everything properly."""
  3. @kissgyorgy kissgyorgy revised this gist May 24, 2015. 1 changed file with 10 additions and 6 deletions.
    16 changes: 10 additions & 6 deletions conftest.py
    View file
    Original file line number Diff line number Diff line change
    @@ -3,13 +3,19 @@
    from myapp.models import BaseModel
    import pytest

    @pytest.fixture(scope='session')
    def engine():
    return create_engine('postgresql://localhost/test_database)

    @pytest.yield_fixture(scope='session')
    def tables(engine):
    BaseModel.metadata.create_all(engine)
    yield
    BaseModel.metadata.drop_all(engine)

    @pytest.yield_fixture
    def dbsession():
    def dbsession(engine, tables):
    """Returns an sqlalchemy session, and after the test tears down everything properly."""
    engine = create_engine('postgresql://localhost/test_database')
    # create the tables before we start the transaction
    BaseModel.metadata.create_all(engine)
    connection = engine.connect()
    # begin the nested transaction
    transacton = connection.begin()
    @@ -21,7 +27,5 @@ def dbsession():
    session.close()
    # roll back the broader transaction
    transacton.rollback()
    # now we can delete the tables and wont hang
    BaseModel.metadata.drop_all(engine)
    # put back the connection to the connection pool
    connection.close()
  4. @kissgyorgy kissgyorgy created this gist May 24, 2015.
    27 changes: 27 additions & 0 deletions conftest.py
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,27 @@
    from sqlalchemy import create_engine
    from sqlalchemy.orm import Session
    from myapp.models import BaseModel
    import pytest


    @pytest.yield_fixture
    def dbsession():
    """Returns an sqlalchemy session, and after the test tears down everything properly."""
    engine = create_engine('postgresql://localhost/test_database')
    # create the tables before we start the transaction
    BaseModel.metadata.create_all(engine)
    connection = engine.connect()
    # begin the nested transaction
    transacton = connection.begin()
    # use the connection with the already started transaction
    session = Session(bind=connection)

    yield session

    session.close()
    # roll back the broader transaction
    transacton.rollback()
    # now we can delete the tables and wont hang
    BaseModel.metadata.drop_all(engine)
    # put back the connection to the connection pool
    connection.close()