feat: add PostgreSQL support for CI tests and update test database configuration (#6)

Author: umbertogriffo

.github/workflows/ci.yaml

@@ -27,6 +27,15 @@ jobs:
   test:
     runs-on: ubuntu-latest
     timeout-minutes: 20
+    services:
+      postgres:
+        image: public.ecr.aws/docker/library/postgres:17
+        env:
+          POSTGRES_USER: develop
+          POSTGRES_PASSWORD: develop_secret
+          POSTGRES_DB: develop
+        ports:
+          - 5432:5432
     steps:
       - name: Checkout
         uses: actions/checkout@v5
@@ -48,8 +57,14 @@ jobs:
       - name: Install the project
         run: uv sync --locked --all-extras --dev
 
-      - name: Run tests
+      - name: Run tests (SQLite)
         env:
           API_KEY: TEST
         run: |
           uv run pytest -s --durations=5 --log-cli-level=INFO --capture=tee-sys -v
+
+      - name: Run tests (Postgres)
+        env:
+          API_KEY: TEST
+        run: |
+          uv run pytest -s --db postgres --durations=5 --log-cli-level=INFO --capture=tee-sys -v

NOTE.md

@@ -0,0 +1,56 @@
+# Note
+
+## Rollback when using pytest
+
+When using `pytest` with SQLModel and a database, it's important to ensure that changes made during tests do not persist and affect other tests.
+
+Here are some strategies to achieve this:
+
+In `conftest.py` create a session that rolls back after each test:
+```python
+    # Start a transaction that will be rolled back after the test
+    connection = engine.connect()
+    transaction = connection.begin()
+    session = Session(bind=connection)
+
+    yield session
+
+    # Rollback all changes made during the test
+    session.close()
+    transaction.rollback()
+    connection.close()
+```
+
+Or [use a nested transaction for better test isolation](https://github.com/fastapi/sqlmodel/discussions/940):
+```python
+import sqlalchemy as sa
+from sqlmodel import Session
+
+    connection = engine.connect()
+    transaction = connection.begin()
+    session = Session(bind=connection)
+
+    nested = connection.begin_nested()
+
+    @sa.event.listens_for(session, "after_transaction_end")
+    def end_savepoint(session, transaction):
+        nonlocal nested
+        if not nested.is_active:
+            nested = connection.begin_nested()
+
+    yield session
+
+    session.close()
+    transaction.rollback()
+    connection.close()
+```
+
+or recreate the entire database schema between tests (recreating the schema is slow):
+```python
+    with Session(engine) as session:
+        yield session
+
+    # For Postgres, clean up all data between tests
+    if db_type == "postgres":
+        command.downgrade(config, "base")
+```

README.md

@@ -43,7 +43,7 @@ Repo Features:
 
 * Python 3.10+
 * uv 0.8.17+
-* PoatgreSQL 18.1+ (if you want to use PostgreSQL as database)
+* PostgreSQL 18.1+ (if you want to use PostgreSQL as database)
 
 ### Install uv
 

src/alembic/versions/2025_12_18_1212-438f1c13aa3b_init.py

@@ -21,6 +21,11 @@
 
 def upgrade() -> None:
     """Upgrade schema."""
+
+    # Example of handling different JSON types based on the database backend
+    # is_postgres = settings.DATABASE_URL.startswith("postgresql")
+    # json_type = postgresql.JSONB() if is_postgres else sa.JSON()
+
     # ### commands auto generated by Alembic - please adjust! ###
     op.create_table(
         "hero",

tests/api/endpoints/test_heroes.py

@@ -101,7 +101,7 @@ def test_delete_hero(session: Session, client_with_db: TestClient):
     assert hero_in_db is None
 
 
-def test_update_hero_rollback_on_error(
+def test_update_hero_rollback_on_commit_error(
     mocker, client_with_db: TestClient, session: Session
 ):
     """Test that updating a hero rolls back on commit error."""

tests/conftest.py

@@ -3,6 +3,7 @@
 from pathlib import Path
 
 import pytest
+import sqlalchemy as sa
 from alembic import command
 from alembic.config import Config
 from api.deps import get_db_session
@@ -11,21 +12,38 @@
 from starlette.testclient import TestClient
 
 
+def pytest_addoption(parser):
+    """Add custom command line options."""
+    parser.addoption(
+        "--db",
+        action="store",
+        default="sqlite",
+        choices=["sqlite", "postgres"],
+        help="Database to use for tests: sqlite (default) or postgres",
+    )
+
+
 @pytest.fixture
 def data_folder_path():
     return Path(__file__).parent.parent / "data"
 
 
 @pytest.fixture(name="session")
-def session_fixture(monkeypatch) -> Session:
+def session_fixture(request, monkeypatch) -> Session:
     """Create a new database session for a test."""
     # TODO: Use an in-memory SQLite database for faster tests if possible.
     #       https://sqlmodel.tiangolo.com/tutorial/fastapi/tests/#memory-database
 
-    # Create a temporary database file
-    fd, path = tempfile.mkstemp(suffix=".db")
-    os.close(fd)
-    db_url = f"sqlite:///{path}"
+    db_type = request.config.getoption("--db")
+
+    if db_type == "postgres":
+        db_url = "postgresql://develop:develop_secret@localhost:5432/develop"
+        path = None
+    else:
+        # Create a temporary database file for SQLite
+        fd, path = tempfile.mkstemp(suffix=".db")
+        os.close(fd)
+        db_url = f"sqlite:///{path}"
 
     # Use monkeypatch to set DATABASE_URL environment variable
     monkeypatch.setattr("config.settings.DATABASE_URL", db_url)
@@ -42,12 +60,31 @@ def session_fixture(monkeypatch) -> Session:
     engine = create_engine(db_url)
     # TODO: Alternatively, you can create tables directly without migrations for simpler setups.
     # create_db_and_tables(engine)
-    with Session(engine) as session:
-        yield session
+
+    # Ensure that changes made during tests do not persist and affect other tests using a nested transaction
+    # This is needed for PostgreSQL since the SQLite is erased after each test by deleting the temp file
+    connection = engine.connect()
+    transaction = connection.begin()
+    session = Session(bind=connection)
+
+    nested = connection.begin_nested()
+
+    @sa.event.listens_for(session, "after_transaction_end")
+    def end_savepoint(session, transaction):
+        nonlocal nested
+        if not nested.is_active:
+            nested = connection.begin_nested()
+
+    yield session
+
+    session.close()
+    transaction.rollback()
+    connection.close()
 
     # Clean up
     engine.dispose()
-    os.unlink(path)
+    if path:
+        os.unlink(path)
 
 
 @pytest.fixture(name="client_with_db")