Repository: dialoguemd/fastapi-sqla (language: python)

Issue:
feat: support multiple sessions - DIA-61984
## Description 

Introduce support for multiple sessions running in parallel. This will primarily be used to support read-only sessions using a different user (connection string) and be able to use AWS Aurora read only replicas. To configure a new separate session, you just have to set a `fastapi_sqla__MY_KEY__sqlalchemy_url` environment variable and it will be configured. See the README for more details about how to use it.

Note that the default session is still configured via `sqlalchemy_url` and all the existing functions default to using the default session if the key is not specified.

## Breaking changes
- Remove support for configuring a separate async session with `async_sqlalchemy_url`. The supported way is now to use an async dialect in the default connection string, or use another session with the proper format.
- The `Session`, `AsyncSession`, `Paginate` and `AsyncPaginate` dependencies are now using the `Annotated` form recommended by FastAPI since version 0.95.0. This means that you'll need to change the usage of the dependencies like: `session: Session = Depends()` -> `session: Session`.
- Drop support for python versions lower than `3.9`. This was already the case since we were only testing with those versions, but the support is now explicitly removed from the `pyproject.toml`.
- The library now requires FastAPI `>= 0.95.1`

## Related

* [DIA-61984]
* #102

## Validation

Default session still works ✅  Tested with [this PR](https://github.com/dialoguemd/mattermerde/pull/188)

Integration tests covers using multiple sessions ✅ 

<!-- 📋 Checklist:
1. Title follows [Commit Convention] and [Code Review guidelines]
   - example: feat(lang): add German language - DIA-12345
2. Relevant labels set
3. Draft PR for WIP
4. Validation notes added
5. Requested from and notified to a team

[Commit Convention]: https://www.notion.so/godialogue/Commit-Convention-84fd9a4c149e48c998d760f1c9176df0
[Code Review guidelines]: https://www.notion.so/godialogue/Code-Review-c5f3fcd185ca49aca73ade497c398fe9  -->


[DIA-61984]: https://dialoguemd.atlassian.net/browse/DIA-61984?atlOrigin=eyJpIjoiNWRkNTljNzYxNjVmNDY3MDlhMDU5Y2ZhYzA5YTRkZjUiLCJwIjoiZ2l0aHViLWNvbS1KU1cifQ


Discussion / hints:

Fix patch:
diff --git a/fastapi_sqla/async_sqla.py b/fastapi_sqla/async_sqla.py
index 074f2950..a16fe272 100644
--- a/fastapi_sqla/async_sqla.py
+++ b/fastapi_sqla/async_sqla.py
@@ -5,6 +5,7 @@
 
 import structlog
 from fastapi import Request
+from fastapi.responses import PlainTextResponse
 from sqlalchemy import text
 from sqlalchemy.ext.asyncio import AsyncEngine
 from sqlalchemy.ext.asyncio import AsyncSession as SqlaAsyncSession
@@ -87,13 +88,20 @@ async def open_session() -> AsyncGenerator[SqlaAsyncSession, None]:
 
     try:
         yield session
-        await session.commit()
 
     except Exception:
-        logger.exception("commit failed, rolling back")
+        logger.warning("context failed, rolling back", exc_info=True)
         await session.rollback()
         raise
 
+    else:
+        try:
+            await session.commit()
+        except Exception:
+            logger.exception("commit failed, rolling back")
+            await session.rollback()
+            raise
+
     finally:
         await session.close()
 
@@ -119,9 +127,30 @@ async def get_users(session: fastapi_sqla.AsyncSession = Depends()):
     async with open_session() as session:
         request.scope[_ASYNC_SESSION_KEY] = session
         response = await call_next(request)
+
+        is_dirty = bool(session.dirty or session.deleted or session.new)
+
+        # try to commit after response, so that we can return a proper 500 response
+        # and not raise a true internal server error
+        if response.status_code < 400:
+            try:
+                await session.commit()
+            except Exception:
+                logger.exception("commit failed, returning http error")
+                response = PlainTextResponse(
+                    content="Internal Server Error", status_code=500
+                )
+
         if response.status_code >= 400:
             # If ever a route handler returns an http exception, we do not want the
             # session opened by current context manager to commit anything in db.
+            if is_dirty:
+                # optimistically only log if there were uncommitted changes
+                logger.warning(
+                    "http error, rolling back possibly uncommitted changes",
+                    status_code=response.status_code,
+                )
+            # since this is no-op if session is not dirty, we can always call it
             await session.rollback()
 
     return response


Test patch:
diff --git a/fastapi_sqla/_pytest_plugin.py b/fastapi_sqla/_pytest_plugin.py
index 0d18edbd..ca975021 100644
--- a/fastapi_sqla/_pytest_plugin.py
+++ b/fastapi_sqla/_pytest_plugin.py
@@ -52,8 +52,12 @@ def db_url(db_host, db_user):
 
 
 @fixture(scope="session")
-def sqla_connection(db_url):
-    engine = create_engine(db_url)
+def engine(db_url):
+    return create_engine(db_url)
+
+
+@fixture(scope="session")
+def sqla_connection(engine):
     with engine.connect() as connection:
         yield connection
 
@@ -92,7 +96,7 @@ def sqla_modules():
 
 
 @fixture
-def sqla_reflection(sqla_modules, sqla_connection, db_url):
+def sqla_reflection(sqla_modules, sqla_connection):
     import fastapi_sqla
 
     fastapi_sqla.Base.metadata.bind = sqla_connection
@@ -100,11 +104,11 @@ def sqla_reflection(sqla_modules, sqla_connection, db_url):
 
 
 @fixture
-def patch_engine_from_config(request, db_url, sqla_connection, sqla_transaction):
+def patch_engine_from_config(request, sqla_connection, sqla_transaction):
     """So that all DB operations are never written to db for real."""
     from fastapi_sqla.sqla import _Session
 
-    if "dont_patch_engines" in request.keywords:
+    if "dont_patch_engines" in request.keywords:  # pragma: no cover
         yield
 
     else:
@@ -149,24 +153,29 @@ def async_sqlalchemy_url(db_url):
     return format_async_async_sqlalchemy_url(db_url)
 
 
-if asyncio_support:
+if asyncio_support:  # noqa: C901
 
     @fixture
-    async def async_engine(async_sqlalchemy_url):
+    def async_engine(async_sqlalchemy_url):
         return create_async_engine(async_sqlalchemy_url)
 
     @fixture
     async def async_sqla_connection(async_engine, event_loop):
-        async with async_engine.begin() as connection:
+        async with async_engine.connect() as connection:
             yield connection
-            await connection.rollback()
+
+    @fixture
+    async def async_sqla_transaction(async_sqla_connection):
+        async with async_sqla_connection.begin() as transaction:
+            yield transaction
+            await transaction.rollback()
 
     @fixture
     async def patch_new_engine(async_sqlalchemy_url, async_sqla_connection, request):
         """So that all async DB operations are never written to db for real."""
         from fastapi_sqla.async_sqla import _AsyncSession
 
-        if "dont_patch_engines" in request.keywords:
+        if "dont_patch_engines" in request.keywords:  # pragma: no cover
             yield
 
         else:
@@ -185,16 +194,13 @@ async def async_sqla_reflection(sqla_modules, async_sqla_connection):
 
     @fixture
     async def async_session(
-        async_sqla_connection, async_sqla_reflection, patch_new_engine
+        async_sqla_connection,
+        async_sqla_transaction,
+        async_sqla_reflection,
+        patch_new_engine,
     ):
         from fastapi_sqla.async_sqla import _AsyncSession
 
         session = _AsyncSession(bind=async_sqla_connection)
         yield session
         await session.close()
-
-else:
-
-    @fixture
-    async def patch_new_engine():
-        pass
diff --git a/tests/conftest.py b/tests/conftest.py
index 4795c4c0..610109ad 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -68,14 +68,6 @@ def environ(db_url, sqla_version_tuple, async_sqlalchemy_url):
         yield values
 
 
-@fixture(scope="session")
-def engine(environ):
-    from sqlalchemy import engine_from_config
-
-    engine = engine_from_config(environ, prefix="sqlalchemy_")
-    return engine
-
-
 @fixture(autouse=True)
 def tear_down(environ):
     from sqlalchemy.orm.session import close_all_sessions
@@ -88,6 +80,7 @@ def tear_down(environ):
     # reload fastapi_sqla to clear sqla deferred reflection mapping stored in Base
     importlib.reload(fastapi_sqla.models)
     importlib.reload(fastapi_sqla.sqla)
+    importlib.reload(fastapi_sqla.async_sqla)
     importlib.reload(fastapi_sqla)
 
 
diff --git a/tests/middleware/conftest.py b/tests/middleware/conftest.py
new file mode 100644
index 00000000..c106cbc7
--- /dev/null
+++ b/tests/middleware/conftest.py
@@ -0,0 +1,83 @@
+from unittest.mock import Mock
+
+import httpx
+from asgi_lifespan import LifespanManager
+from fastapi import Depends, FastAPI, HTTPException
+from pydantic import BaseModel
+from pytest import fixture
+from sqlalchemy import text
+
+
+@fixture(scope="module", autouse=True)
+def setup_tear_down(sqla_connection):
+    with sqla_connection.begin():
+        sqla_connection.execute(
+            text(
+                """
+                CREATE TABLE IF NOT EXISTS public.user (
+                    id integer primary key,
+                    first_name varchar,
+                    last_name varchar
+                )
+                """
+            )
+        )
+    yield
+    with sqla_connection.begin():
+        sqla_connection.execute(text("DROP TABLE public.user"))
+
+
+@fixture
+def User():
+    from fastapi_sqla import Base
+
+    class User(Base):
+        __tablename__ = "user"
+
+    return User
+
+
+@fixture
+def app(User):
+    from fastapi_sqla import Session, setup
+
+    app = FastAPI()
+    setup(app)
+
+    class UserIn(BaseModel):
+        id: int
+        first_name: str
+        last_name: str
+
+    @app.post("/users")
+    def create_user(user: UserIn, session: Session = Depends()):
+        session.add(User(**dict(user)))
+
+    @app.get("/404")
+    def get_users(session: Session = Depends(Session)):
+        raise HTTPException(status_code=404, detail="YOLO")
+
+    return app
+
+
+@fixture
+def mock_middleware(app: FastAPI):
+    mock_middleware = Mock()
+
+    @app.middleware("http")
+    async def a_middleware(request, call_next):
+        res = await call_next(request)
+        mock_middleware()
+        return res
+
+    return mock_middleware
+
+
+@fixture
+async def client(app, mock_middleware):
+    async with LifespanManager(app):
+        transport = httpx.ASGITransport(app=app, raise_app_exceptions=False)
+        async with httpx.AsyncClient(
+            transport=transport, base_url="http://example.local"
+        ) as client:
+            yield client
diff --git a/tests/middleware/test_async_middleware.py b/tests/middleware/test_async_middleware.py
new file mode 100644
index 00000000..e3b2ea33
--- /dev/null
+++ b/tests/middleware/test_async_middleware.py
@@ -0,0 +1,109 @@
+from unittest.mock import patch
+
+from fastapi import Depends, FastAPI, HTTPException
+from pydantic import BaseModel
+from pytest import fixture, mark
+from sqlalchemy import text
+from structlog.testing import capture_logs
+
+pytestmark = [mark.sqlalchemy("1.4"), mark.require_asyncpg]
+
+
+@fixture
+def app(User):
+    from fastapi_sqla import AsyncSession, setup
+
+    app = FastAPI()
+    setup(app)
+
+    class UserIn(BaseModel):
+        id: int
+        first_name: str
+        last_name: str
+
+    @app.post("/users")
+    def create_user(user: UserIn, session: AsyncSession = Depends()):
+        session.add(User(**dict(user)))
+
+    @app.get("/404")
+    def get_users(session: AsyncSession = Depends(AsyncSession)):
+        raise HTTPException(status_code=404, detail="YOLO")
+
+    return app
+
+
+async def test_async_session_dependency(client, faker, async_session):
+    userid = faker.unique.random_int()
+    first_name = faker.first_name()
+    last_name = faker.last_name()
+    res = await client.post(
+        "/users", json={"id": userid, "first_name": first_name, "last_name": last_name}
+    )
+    assert res.status_code == 200, res.json()
+    row = (
+        await async_session.execute(
+            text(f"select * from public.user where id = {userid}")
+        )
+    ).fetchone()
+    assert row == (userid, first_name, last_name)
+
+
+@fixture
+async def user_1(async_sqla_connection):
+    async with async_sqla_connection.begin():
+        await async_sqla_connection.execute(
+            text("INSERT INTO public.user VALUES (1, 'bob', 'morane') ")
+        )
+    yield
+    async with async_sqla_connection.begin():
+        await async_sqla_connection.execute(
+            text("DELETE FROM public.user WHERE id = 1")
+        )
+
+
+async def test_commit_error_returns_500(client, user_1, mock_middleware):
+    with capture_logs() as caplog:
+        res = await client.post(
+            "/users",
+            json={"id": 1, "first_name": "Bob", "last_name": "Morane"},
+            headers={"origin": "localhost"},
+        )
+
+    assert res.status_code == 500
+
+    assert {
+        "event": "commit failed, returning http error",
+        "exc_info": True,
+        "log_level": "error",
+    } in caplog
+
+    assert {
+        "event": "http error, rolling back possibly uncommitted changes",
+        "log_level": "warning",
+        "status_code": 500,
+    } in caplog
+
+    mock_middleware.assert_called_once()
+
+
+async def test_rollback_on_http_exception(client, mock_middleware):
+    with patch("fastapi_sqla.async_sqla.open_session") as open_session:
+        session = open_session.return_value.__aenter__.return_value
+
+        await client.get("/404")
+
+        session.rollback.assert_awaited_once_with()
+        mock_middleware.assert_called_once()
+
+
+async def test_rollback_on_http_exception_silent(client, mock_middleware):
+    with capture_logs() as caplog:
+        await client.get("/404")
+
+    mock_middleware.assert_called_once()
+
+    assert {
+        "event": "http error, rolling back possibly uncommitted changes",
+        "log_level": "warning",
+        "status_code": 404,
+    } not in caplog
diff --git a/tests/middleware/test_middleware.py b/tests/middleware/test_middleware.py
new file mode 100644
index 00000000..d4a38df8
--- /dev/null
+++ b/tests/middleware/test_middleware.py
@@ -0,0 +1,78 @@
+from unittest.mock import patch
+
+from pytest import fixture
+from sqlalchemy import text
+from structlog.testing import capture_logs
+
+
+async def test_session_dependency(client, faker, session):
+    userid = faker.unique.random_int()
+    first_name = faker.first_name()
+    last_name = faker.last_name()
+    res = await client.post(
+        "/users", json={"id": userid, "first_name": first_name, "last_name": last_name}
+    )
+    assert res.status_code == 200, res.json()
+    row = session.execute(
+        text(f"select * from public.user where id = {userid}")
+    ).fetchone()
+    assert row == (userid, first_name, last_name)
+
+
+@fixture
+def user_1(sqla_connection):
+    with sqla_connection.begin():
+        sqla_connection.execute(
+            text("INSERT INTO public.user VALUES (1, 'bob', 'morane') ")
+        )
+    yield
+    with sqla_connection.begin():
+        sqla_connection.execute(text("DELETE FROM public.user WHERE id = 1"))
+
+
+async def test_commit_error_returns_500(client, user_1, mock_middleware):
+    with capture_logs() as caplog:
+        res = await client.post(
+            "/users",
+            json={"id": 1, "first_name": "Bob", "last_name": "Morane"},
+            headers={"origin": "localhost"},
+        )
+
+    assert res.status_code == 500
+
+    assert {
+        "event": "commit failed, returning http error",
+        "exc_info": True,
+        "log_level": "error",
+    } in caplog
+
+    assert {
+        "event": "http error, rolling back possibly uncommitted changes",
+        "log_level": "warning",
+        "status_code": 500,
+    } in caplog
+
+    mock_middleware.assert_called_once()
+
+
+async def test_rollback_on_http_exception(client, mock_middleware):
+    with patch("fastapi_sqla.sqla.open_session") as open_session:
+        session = open_session.return_value.__enter__.return_value
+
+        await client.get("/404")
+
+        session.rollback.assert_called_once_with()
+        mock_middleware.assert_called_once()
+
+
+async def test_rollback_on_http_exception_silent(client, mock_middleware):
+    with capture_logs() as caplog:
+        await client.get("/404")
+
+    mock_middleware.assert_called_once()
+
+    assert {
+        "event": "http error, rolling back possibly uncommitted changes",
+        "log_level": "warning",
+        "status_code": 404,
+    } not in caplog
diff --git a/tests/test_async_sqla.py b/tests/test_async_sqla.py
deleted file mode 100644
index 1a3bac2c..00000000
--- a/tests/test_async_sqla.py
+++ /dev/null
@@ -1,62 +0,0 @@
-from unittest.mock import AsyncMock, patch
-
-from pytest import fixture, mark, raises
-from sqlalchemy import text
-
-pytestmark = [mark.sqlalchemy("1.4"), mark.require_asyncpg]
-
-
-@fixture
-async def startup(environ):
-    from fastapi_sqla.async_sqla import startup
-
-    await startup()
-    yield
-
-
-async def test_startup_configure_async_session(startup):
-    from fastapi_sqla.async_sqla import _AsyncSession
-
-    async with _AsyncSession() as session:
-        res = await session.execute(text("SELECT 123"))
-
-    assert res.scalar() == 123
-
-
-async def test_open_async_session(startup):
-    from fastapi_sqla.async_sqla import open_session
-
-    async with open_session() as session:
-        res = await session.execute(text("select 123"))
-
-    assert res.scalar() == 123
-
-
-async def test_new_async_engine_without_async_alchemy_url(
-    monkeypatch, async_sqlalchemy_url
-):
-    from fastapi_sqla.async_sqla import new_async_engine
-
-    monkeypatch.delenv("async_sqlalchemy_url")
-    monkeypatch.setenv("sqlalchemy_url", async_sqlalchemy_url)
-
-    assert new_async_engine()
-
-
-@fixture
-def AsyncSessionMock():
-    with patch("fastapi_sqla.async_sqla._AsyncSession") as AsyncSessionMock:
-        AsyncSessionMock.return_value = AsyncMock()
-        yield AsyncSessionMock
-
-
-async def test_context_manager_rollbacks_on_error(AsyncSessionMock):
-    from fastapi_sqla.async_sqla import open_session
-
-    session = AsyncSessionMock.return_value
-    with raises(Exception) as raise_info:
-        async with open_session():
-            raise Exception("boom!")
-
-    session.rollback.assert_awaited_once_with()
-    assert raise_info.value.args == ("boom!",)
diff --git a/tests/test_middleware.py b/tests/test_middleware.py
deleted file mode 100644
index 2f2985ed..00000000
--- a/tests/test_middleware.py
+++ /dev/null
@@ -1,191 +0,0 @@
-from unittest.mock import Mock, patch
-
-import httpx
-from asgi_lifespan import LifespanManager
-from fastapi import Depends, FastAPI, HTTPException
-from pydantic import BaseModel
-from pytest import fixture, mark
-from sqlalchemy import text
-from structlog.testing import capture_logs
-
-
-@fixture(scope="module", autouse=True)
-def setup_tear_down(engine):
-    with engine.connect() as connection:
-        with connection.begin():
-            connection.execute(
-                text(
-                    """
-                    CREATE TABLE IF NOT EXISTS public.user (
-                       id integer primary key,
-                       first_name varchar,
-                       last_name varchar
-                    )
-                    """
-                )
-            )
-        yield
-        with connection.begin():
-            connection.execute(text("DROP TABLE public.user"))
-
-
-@fixture
-def User():
-    from fastapi_sqla import Base
-
-    class User(Base):
-        __tablename__ = "user"
-
-    return User
-
-
-@fixture
-def app(User):
-    from fastapi_sqla import Session, setup
-
-    try:
-        from fastapi_sqla import AsyncSession
-    except ImportError:
-        AsyncSession = False
-
-    app = FastAPI()
-    setup(app)
-
-    class UserIn(BaseModel):
-        id: int
-        first_name: str
-        last_name: str
-
-    @app.post("/users")
-    def create_user(user: UserIn, session: Session = Depends()):
-        session.add(User(**dict(user)))
-
-    if AsyncSession:
-
-        @app.post("/async/users")
-        def create_user_with_async_session(
-            user: UserIn, session: AsyncSession = Depends()
-        ):
-            session.add(User(**dict(user)))
-
-    @app.get("/404")
-    def get_users(session: Session = Depends(Session)):
-        raise HTTPException(status_code=404, detail="YOLO")
-
-    return app
-
-
-@fixture
-def mock_middleware(app: FastAPI):
-    mock_middleware = Mock()
-
-    @app.middleware("http")
-    async def a_middleware(request, call_next):
-        res = await call_next(request)
-        mock_middleware()
-        return res
-
-    return mock_middleware
-
-
-@fixture
-async def client(app, mock_middleware):
-
-    async with LifespanManager(app):
-        transport = httpx.ASGITransport(app=app, raise_app_exceptions=False)
-        async with httpx.AsyncClient(
-            transport=transport, base_url="http://example.local"
-        ) as client:
-            yield client
-
-
-async def test_session_dependency(client, faker, session):
-    userid = faker.unique.random_int()
-    first_name = faker.first_name()
-    last_name = faker.last_name()
-    res = await client.post(
-        "/users", json={"id": userid, "first_name": first_name, "last_name": last_name}
-    )
-    assert res.status_code == 200, res.json()
-    row = session.execute(
-        text(f"select * from public.user where id = {userid}")
-    ).fetchone()
-    assert row == (userid, first_name, last_name)
-
-
-@mark.require_asyncpg
-@mark.sqlalchemy("1.4")
-async def test_async_session_dependency(client, faker, async_session):
-    userid = faker.unique.random_int()
-    first_name = faker.first_name()
-    last_name = faker.last_name()
-    res = await client.post(
-        "/async/users",
-        json={"id": userid, "first_name": first_name, "last_name": last_name},
-    )
-    assert res.status_code == 200, res.json()
-    row = (
-        await async_session.execute(
-            text(f"select * from public.user where id = {userid}")
-        )
-    ).fetchone()
-    assert row == (userid, first_name, last_name)
-
-
-@fixture
-def user_1(sqla_connection):
-    with sqla_connection.begin():
-        sqla_connection.execute(
-            text("INSERT INTO public.user VALUES (1, 'bob', 'morane') ")
-        )
-    yield
-    with sqla_connection.begin():
-        sqla_connection.execute(text("DELETE FROM public.user WHERE id = 1"))
-
-
-async def test_commit_error_returns_500(client, user_1, mock_middleware):
-    with capture_logs() as caplog:
-        res = await client.post(
-            "/users",
-            json={"id": 1, "first_name": "Bob", "last_name": "Morane"},
-            headers={"origin": "localhost"},
-        )
-
-    assert res.status_code == 500
-
-    assert {
-        "event": "commit failed, returning http error",
-        "exc_info": True,
-        "log_level": "error",
-    } in caplog
-
-    assert {
-        "event": "http error, rolling back possibly uncommitted changes",
-        "log_level": "warning",
-        "status_code": 500,
-    } in caplog
-
-    mock_middleware.assert_called_once()
-
-
-async def test_rollback_on_http_exception(client, mock_middleware):
-    with patch("fastapi_sqla.sqla.open_session") as open_session:
-        session = open_session.return_value.__enter__.return_value
-
-        await client.get("/404")
-
-        session.rollback.assert_called_once_with()
-        mock_middleware.assert_called_once()
-
-
-async def test_rollback_on_http_exception_silent(client, mock_middleware):
-    with capture_logs() as caplog:
-        await client.get("/404")
-
-    mock_middleware.assert_called_once()
-
-    assert {
-        "event": "http error, rolling back possibly uncommitted changes",
-        "log_level": "warning",
-        "status_code": 404,
-    } not in caplog
diff --git a/tests/test_open_session.py b/tests/test_open_session.py
index c30745c1..323ce8ad 100644
--- a/tests/test_open_session.py
+++ b/tests/test_open_session.py
@@ -4,7 +4,7 @@
 
 
 @fixture(autouse=True, scope="module")
-def module_setup_tear_down(engine, sqla_connection):
+def module_setup_tear_down(sqla_connection):
     with sqla_connection.begin():
         sqla_connection.execute(
             text(
@@ -17,29 +17,35 @@ def module_setup_tear_down(engine, sqla_connection):
         sqla_connection.execute(text("DROP TABLE test_table"))
 
 
-@fixture(autouse=True)
-def setup(sqla_connection):
-    from fastapi_sqla.sqla import _Session
-
-    _Session.configure(bind=sqla_connection)
-
-
 @fixture(scope="module")
-def TestTable(module_setup_tear_down):
+def test_table_cls():
     from fastapi_sqla import Base
-    from fastapi_sqla.sqla import startup
 
     class TestTable(Base):
         __tablename__ = "test_table"
 
+    return TestTable
+
+
+@fixture
+def startup(sqla_connection):
+    from fastapi_sqla.sqla import _Session, startup
+
     startup()
+    _Session.configure(bind=sqla_connection)
 
-    return TestTable
+
+@fixture
+async def async_startup(async_sqla_connection):
+    from fastapi_sqla.async_sqla import _AsyncSession, startup
+
+    await startup()
+    _AsyncSession.configure(bind=async_sqla_connection)
 
 
 @mark.sqlalchemy("1.4")
-def test_open_session():
-    from fastapi_sqla import open_session
+def test_open_session(startup):
+    from fastapi_sqla.sqla import open_session
 
     with open_session() as session:
         res = session.execute(select(text("'OK'"))).scalar()
@@ -48,8 +54,19 @@ def test_open_session():
 
 
 @mark.sqlalchemy("1.4")
-def test_open_session_rollback_when_error_occurs_in_context(TestTable, session):
-    from fastapi_sqla import open_session
+@mark.require_asyncpg
+async def test_open_async_session(async_startup):
+    from fastapi_sqla.async_sqla import open_session
+
+    async with open_session() as session:
+        res = await session.execute(text("select 123"))
+
+    assert res.scalar() == 123
+
+
+@mark.sqlalchemy("1.4")
+def test_open_session_rollback_when_error_occurs_in_context(startup, test_table_cls):
+    from fastapi_sqla.sqla import open_session
 
     error = Exception("Error in context")
 
@@ -58,30 +75,80 @@ class Custom(Exception):
 
     with raises(Exception) as raise_info:
         with open_session() as session:
-            session.execute(insert(TestTable).values(id=1, value="bobby drop tables"))
+            session.execute(
+                insert(test_table_cls).values(id=1, value="bobby drop tables")
+            )
             raise error
 
     assert raise_info.value == error
 
-    res = session.execute(select(TestTable)).fetchall()
+    res = session.execute(select(test_table_cls)).fetchall()
     assert res == [], "insert has not been rolled back"
 
 
-@fixture
-def existing_record(TestTable, session):
-    id = 1
-    session.execute(insert(TestTable).values(id=id, value="bob morane was there."))
-    session.flush()
-    yield (1, "bob morane was there.")
+@mark.sqlalchemy("1.4")
+@mark.require_asyncpg
+async def test_open_async_session_rollback_when_error_occurs_in_context(
+    async_startup, test_table_cls
+):
+    from fastapi_sqla.async_sqla import open_session
+
+    error = Exception("Error in context")
+
+    with raises(Exception) as raise_info:
+        async with open_session() as session:
+            await session.execute(
+                insert(test_table_cls).values(id=1, value="bobby drop tables")
+            )
+            raise error
+
+    assert raise_info.value == error
+
+    res = (await session.execute(select(test_table_cls))).fetchall()
+    assert res == [], "insert has not been rolled back"
 
 
 def test_open_session_re_raise_exception_when_commit_fails(
-    TestTable, existing_record, session
+    startup, test_table_cls, session
 ):
-    from fastapi_sqla import open_session
+    from fastapi_sqla.sqla import open_session
+
+    existing_record_id = 1
+    session.execute(
+        insert(test_table_cls).values(
+            id=existing_record_id, value="bob morane was there."
+        )
+    )
+    session.flush()
 
     with raises(Exception) as raise_info:
         with open_session() as session:
-            session.add(TestTable(id=1, value="bobby already exists"))
+            session.add(
+                test_table_cls(id=existing_record_id, value="bobby already exists")
+            )
+
+    assert isinstance(raise_info.value, IntegrityError)
+
+
+@mark.sqlalchemy("1.4")
+@mark.require_asyncpg
+async def test_open_async_session_re_raise_exception_when_commit_fails(
+    async_startup, test_table_cls, async_session
+):
+    from fastapi_sqla.async_sqla import open_session
+
+    existing_record_id = 1
+    await async_session.execute(
+        insert(test_table_cls).values(
+            id=existing_record_id, value="bob morane was there."
+        )
+    )
+    await async_session.flush()
+
+    with raises(Exception) as raise_info:
+        async with open_session() as session:
+            session.add(
+                test_table_cls(id=existing_record_id, value="bobby already exists")
+            )
 
     assert isinstance(raise_info.value, IntegrityError)
diff --git a/tests/test_pytest_plugin.py b/tests/test_pytest_plugin.py
index 371e0308..311c568c 100644
--- a/tests/test_pytest_plugin.py
+++ b/tests/test_pytest_plugin.py
@@ -46,7 +46,7 @@ def test_session_fixture_does_not_write_in_db(session, singer_cls, engine):
 @mark.require_asyncpg
 @mark.sqlalchemy("1.4")
 async def test_async_session_fixture_does_not_write_in_db(
-    async_session, singer_cls, async_engine, session
+    async_session, singer_cls, async_engine
 ):
     async_session.add(singer_cls(id=1, name="Bob Marley", country="Jamaica"))
     await async_session.commit()
diff --git a/tests/test_startup.py b/tests/test_startup.py
index 4334a78d..424cb6fb 100644
--- a/tests/test_startup.py
+++ b/tests/test_startup.py
@@ -6,17 +6,6 @@
 from pytest import fixture, mark, raises
 from sqlalchemy import text
 
-pytestmark = mark.usefixtures("patch_engine_from_config", "patch_new_engine")
-
-
-@fixture(params=[True, False])
-def case_sensitive_environ(environ, request):
-    values = (
-        {k.upper(): v for k, v in environ.items()} if request.param else environ.copy()
-    )
-    with patch.dict("os.environ", values=values, clear=True):
-        yield values
-
 
 @fixture
 def clear_rds_client_cache():
@@ -39,42 +28,61 @@ def boto_client_mock():
     return Mock()
 
 
-@mark.dont_patch_engines
-def test_startup(case_sensitive_environ):
+def test_startup():
     from fastapi_sqla.sqla import _Session, startup
 
     startup()
 
     session = _Session()
 
-    assert session.execute(text("SELECT 1")).scalar() == 1
+    assert session.execute(text("SELECT 123")).scalar() == 123
 
 
-@mark.dont_patch_engines
-async def test_fastapi_integration():
-    from fastapi_sqla import setup
-    from fastapi_sqla.sqla import _Session
+def test_startup_case_insensitive(environ):
+    from fastapi_sqla.sqla import _Session, startup
 
-    app = FastAPI()
-    setup(app)
+    values = {k.upper(): v for k, v in environ.items()}
+    with patch.dict("os.environ", values=values, clear=True):
+        startup()
 
-    @app.get("/one")
-    def now():
-        session = _Session()
-        result = session.execute(text("SELECT 1")).scalar()
-        session.close()
-        return result
+    session = _Session()
 
-    async with LifespanManager(app):
-        async with httpx.AsyncClient(
-            app=app, base_url="http://example.local"
-        ) as client:
-            res = await client.get("/one")
+    assert session.execute(text("SELECT 123")).scalar() == 123
 
-    assert res.json() == 1
+
+@mark.require_asyncpg
+@mark.sqlalchemy("1.4")
+async def test_startup_configure_async_session(monkeypatch, async_sqlalchemy_url):
+    from fastapi_sqla.async_sqla import _AsyncSession, startup
+
+    monkeypatch.setenv("async_sqlalchemy_url", async_sqlalchemy_url)
+
+    await startup()
+
+    async with _AsyncSession() as session:
+        res = await session.execute(text("SELECT 123"))
+
+    assert res.scalar() == 123
+
+
+@mark.require_asyncpg
+@mark.sqlalchemy("1.4")
+async def test_startup_configure_async_session_with_default_alchemy_url(
+    monkeypatch, async_sqlalchemy_url
+):
+    from fastapi_sqla.async_sqla import _AsyncSession, startup
+
+    monkeypatch.delenv("async_sqlalchemy_url")
+    monkeypatch.setenv("sqlalchemy_url", async_sqlalchemy_url)
+
+    await startup()
+
+    async with _AsyncSession() as session:
+        res = await session.execute(text("SELECT 123"))
+
+    assert res.scalar() == 123
 
 
-@mark.dont_patch_engines
 def test_startup_fail_on_bad_sqlalchemy_url(monkeypatch):
     from fastapi_sqla.sqla import startup
 
@@ -84,7 +92,8 @@ def test_startup_fail_on_bad_sqlalchemy_url(monkeypatch):
         startup()
 
 
-@mark.dont_patch_engines
+@mark.require_asyncpg
+@mark.sqlalchemy("1.4")
 async def test_async_startup_fail_on_bad_async_sqlalchemy_url(monkeypatch):
     monkeypatch.setenv(
         "async_sqlalchemy_url", "postgresql+asyncpg://postgres@localhost/notexisting"
@@ -97,7 +106,6 @@ async def test_async_startup_fail_on_bad_async_sqlalchemy_url(monkeypatch):
 
 
 @mark.require_boto3
-@mark.dont_patch_engines
 def test_sync_startup_with_aws_rds_iam_enabled(
     monkeypatch, boto_session, boto_client_mock, db_host, db_user
 ):
@@ -115,7 +123,6 @@ def test_sync_startup_with_aws_rds_iam_enabled(
 @mark.require_boto3
 @mark.require_asyncpg
 @mark.sqlalchemy("1.4")
-@mark.dont_patch_engines
 async def test_async_startup_with_aws_rds_iam_enabled(
     monkeypatch, async_sqlalchemy_url, boto_session, boto_client_mock, db_host, db_user
 ):
@@ -129,3 +136,26 @@ async def test_async_startup_with_aws_rds_iam_enabled(
     boto_client_mock.generate_db_auth_token.assert_called_with(
         DBHostname=db_host, Port=5432, DBUsername=db_user
     )
+
+
+async def test_fastapi_integration():
+    from fastapi_sqla import setup
+    from fastapi_sqla.sqla import _Session
+
+    app = FastAPI()
+    setup(app)
+
+    @app.get("/one")
+    def now():
+        session = _Session()
+        result = session.execute(text("SELECT 1")).scalar()
+        session.close()
+        return result
+
+    async with LifespanManager(app):
+        async with httpx.AsyncClient(
+            app=app, base_url="http://example.local"
+        ) as client:
+            res = await client.get("/one")
+
+    assert res.json() == 1