SQLalchemy - How to yield two database connections when running test cases

Question

Below is database.py file:

def get_dba_pgdsn() -> str:
    return f"mysql://{user}:{pass}@dba"


def get_dbb_pgdsn() -> str:
    return f"mysql://{user}:{pass}@dbb"


def engines() -> Engine:
    engines = {
        "dba": get_dba_pgdsn(),
        "dbb": get_dbb_pgdsn()
    }
    return engines


def get_engine() -> Engine:
    if ENGINE is None:
        raise Exception(
            "SQLAlchemy DB Engine has not been created yet. "
            "This is normally done in database.init()"
        )
    return ENGINE


Base = declarative_base(cls=SQLAlchemyBase)

engine_list = engines()

Session_a = scoped_session(sessionmaker(engine_list["dba"]))
Session_b = scoped_session(sessionmaker(engine_list["dbb"]))


def init(session_expire_on_commit=True):
    global ENGINE

    if not ENGINE:
        ENGINE = engines()
        Session_a.configure(bind=engine_list["dba"], expire_on_commit=session_expire_on_commit)
        Session_b.configure(bind=engine_list["dbb"], expire_on_commit=session_expire_on_commit)

And this is the conftest file I have:

from shared import database
from sqlalchemy import text
import pytest


@pytest.fixture()
def db_session_a():
    
    database.init()
    yield database.Session_a
    database.Session_a.rollback()
    for table in reversed(database.Base.metadata.sorted_tables):
        if table.name != "flyway_schema_history":
            sql_stmt = text(f"TRUNCATE TABLE {table.name} RESTART IDENTITY CASCADE;")
            database.Session_a.execute(sql_stmt)
    database.Session_a.commit()
    database.Session_a.close_all()


@pytest.fixture()
def db_session_b():
    
    database.init()
    yield database.Session_b
    database.Session_b.rollback()
    for table in reversed(database.Base.metadata.sorted_tables):
        if table.name != "flyway_schema_history":
            sql_stmt = text(f"TRUNCATE TABLE {table.name} RESTART IDENTITY CASCADE;")
            database.Session_b.execute(sql_stmt)
    database.Session_b.commit()
    database.Session_b.close_all()

And finally the test case

def test_run(self, db_session_a, db_session_b):
    setup_sql_db()
    
    entry = factories.entry(id=1)
    
    print("entry:" + str(entry.id))

I get this error when running

self = <sqlalchemy.orm.session.SessionTransaction object at 0x147f27310>
bind = "mysql://{user}:{pass}@dba"
execution_options = None

def _connection_for_bind(self, bind, execution_options):
    self._assert_active()

    if bind in self._connections:
        if execution_options:
            util.warn(
                "Connection is already established for the "
                "given bind; execution_options ignored"
            )
        return self._connections[bind][0]

    local_connect = False
    should_commit = True

    if self._parent:
        conn = self._parent._connection_for_bind(bind, execution_options)
        if not self.nested:
            return conn
    else:
        if isinstance(bind, engine.Connection):
            conn = bind
            if conn.engine in self._connections:
                raise sa_exc.InvalidRequestError(
                    "Session already has a Connection associated for the "
                    "given Connection's Engine"
                )
        else:
            conn = bind.connect()
            AttributeError: 'str' object has no attribute 'connect'

So my question is given me code above, how can I properly bind these two database connections to be up during the test case execution?

I want to be able to have both database connections up at the same time as I will have test cases that need to execute queries from both databases at the same time.

Maybe this solution for multiple SQLAlchemy databases might help: stackoverflow.com/questions/28238785/… — spaniard
– spaniard, Commented Sep 1, 2023 at 23:50

Lie Ryan · Accepted Answer · 2023-09-09 07:02:45Z

1

The issue you're having is that your engines() function is not returning a dict of engines, it's returning a dict of connection string:

def engines() -> Engine:
    engines = {
        "dba": get_dba_pgdsn(),
        "dbb": get_dbb_pgdsn()
    }
    return engines

Session.configure(bind=...) takes an Engine or a Connection, not a connection string.

Try this instead:

def engines() -> dict[str, Engine]:
    engines = {
        "dba": create_engine(get_dba_pgdsn()),
        "dbb": create_engine(get_dbb_pgdsn()),
    }
    return engines

answered Sep 9, 2023 at 7:02

Lie Ryan

65.3k14 gold badges103 silver badges151 bronze badges

Sign up to request clarification or add additional context in comments.

Collectives™ on Stack Overflow

SQLalchemy - How to yield two database connections when running test cases

1 Answer 1

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related