How To Test Database Transactions With Pytest And SQLModel

Posted on 2024-01-25 Edited on 2024-04-25 In Database Testing Word count in article: 4.7k Reading time ≈ 17 mins.

Almost every backend system uses a database today. After all, you need to persist system and user information.

But with that comes a challenge, your code needs to be able to efficiently read and write from a database. And it’s your job to thoroughly test it.

Databases are capable of smoothly handling read/writes, transactions and rollbacks.

So how do you test your interactions with the database?

How do you leverage the inbuilt database features to write robust I/O operations?

What about handling errors, rollbacks, transactions or network failures? Can you test for these?

Yes, the answer is Pytest, and its unique high-performing capabilities to interact with ORM systems (Object Relation Mapping) like SQLAlchemy and SQLModel.

In this article, I’ll teach and show you how to test database operations (create, read, update, delete) using SQLModel, SQLAlchemy and Pytest.

We’ll explore a practical example, design test cases and discuss how to handle engine and session failures.

You’ll learn how to smartly leverage Pytest hooks and fixtures to efficiently share object instances and code across your test suite with minimal duplication.

Are you excited? Let’s begin.

If you prefer to dive into the code, here’s a link to the example code.

What You’ll Learn

In this article, you’ll learn

How to use ORM tools (like SQLModel and SQLAlchemy) to interact with a database using pure Python (No raw SQL)
How to use Pydantic to validate your database models.
How to write unit tests for core database operations — Create, Read, Update, Delete and Exceptions (connectivity issues, session failures)
How to use Pytest fixtures and hooks in a practical setting to share data across the unit tests using conftest.py.
Using in-memory DBs
Best practices for Database Testing using Pytest

Raw SQL vs ORM tools

When interacting with a database, you have two primary choices: using Raw SQL or ORM (Object-Relational Mapping) tools.

Raw SQL involves writing database queries in SQL, offering you precise control and the power to optimize complex queries.

It’s great if you are comfortable with SQL and require specific database interactions.

You often run these queries ad-hoc via database clients like DBeaver and so on.

On the other hand, ORM tools, like SQLAlchemy or SQLModel (built off SQLAlchemy with Pydantic), abstract the low-level database interactions into object-oriented code.

Thus making it a pleasure to write easy-to-read, well-structured, strongly typed code that is geared towards automation.

API Clients like Flask and Django also have their versions of SQLAlchemy or other ORM tools with an API sauce so once you’re familiar with the concept and fundamentals, it’s worth exploring those.

Before we move on to any example code, let’s understand why you need to test DB operations. If it works on the first instance, that’s good enough, right? Right?

Why Test Database Operations?

Perhaps you’re wondering, why should you even test your database operations.

If your code works that’s good, right?

No, it’s not.

While it may work for the most basic scenario, does it work when called in different sequences with different inputs?

Does it handle errors gracefully?

Can it identify inconsistencies in the data and validate it against the expected database model efficiently?

While you may answer YES to some of these questions, it’s useful to test your code in line with TDD best practices.

It not only helps you identify inevitable shortcomings in your code but also helps you better structure your code, make it easily callable and ensure isolation of key objects and functions.

It helps you think about the working of your code from a 3rd perspective and keeps it maintainable and you accountable.

Not to mention proactive CI testing on pull requests to make sure current functionality still works (called Regression Testing).

This proactive approach saves time, and resources and also builds confidence in your application’s stability and performance.

Enough chit-chat let’s get our hands dirty with a real practical example, one that you’d code in your job.

Prerequisites

To follow this guide, you should have basic working knowledge of:

Python
Pytest
Basics of databases like Primary Key and Datatypes

Set-Up

First, set up your local environment.

Clone the repo here.

.  
├── .gitignore
├── README.md
├── pytest.ini
├── requirements.txt
├── task_manager
│   ├── __init__.py
│   ├── db.py
│   ├── exceptions.py
│   ├── logger.py
│   ├── main.py
│   └── model.py
└── tests
    ├── __init__.py
    ├── conftest.py
    ├── test_db_core_crud.py
    ├── test_db_engine.py
    └── test_db_exceptions.py

Our example code is a simple task_manager application that writes data into a tasks table in a SQL database.

For the sake of simplicity so you can focus on unit tests I’ve used an in-memory database (SQLite) but the same principles apply if you’re working with SQL Server, Postgres or another relational database.

Create a virtual environment and install any requirements (packages) using

1	$ pip install -r requirements.txt

Source Code

Let’s quickly take a look at the example source code.

Our source code contains 2 important files — db.py and model.py .

There are a couple of helper files like exception.py and logger.py .

task_manager/model.py

"""  
Model for the Task Manager  
"""  
from typing import Optional  
from sqlmodel import Field, SQLModel  
from enum import Enum  
from datetime import datetime, timezone  
  
  
class TaskStatus(str, Enum):  
    NOT_STARTED = "Not Started"  
    IN_PROGRESS = "In Progress"  
    COMPLETED = "Completed"  
  
  
class Tasks(SQLModel, table=True):  
    id: Optional[int] = Field(  
        default=None, primary_key=True, description="The ID of the task"  
    )  
    title: str = Field(description="The title of the task")  
    description: Optional[str] = Field(description="The description of the task")  
    status: TaskStatus = Field(  
        sa_column_kwargs={"default": TaskStatus.NOT_STARTED},  
        description="The status of the task",  
    )  
    created_at: datetime = Field(  
        default=datetime.now(timezone.utc),  
        nullable=False,  
        description="The timestamp of when the task was created",  
    )  
    updated_at: datetime = Field(  
        default_factory=datetime.now,  
        nullable=False,  
        description="The timestamp of when the task was updated",  
    )

The above code uses SQLModel to define an object Tasks .

Tasks contains 6 fields

id — The unique ID and primary key for the table
title — Title of a task, string
description — Description of a task, string
status — The Task Status, enum containing values NOT_STARTED , IN_PROGRESS and COMPLETED
created_at — Timestamp when the task was created, datetime
updated_at — Timestamp when the task was updated, datetime

This is a very simple model and in real life, your Entity Relation Diagram (ERD) can be more complex with several normalized tables.

But the concept is largely the same.

Now let’s look at the core of this operation — the database handler.

task_manager/db.py

"""  
Database Operations for the Task Manager  
"""  
import os  
from typing import List  
from datetime import datetime, timezone  
from sqlmodel import SQLModel, create_engine, Session, select  
from task_manager.model import Tasks, TaskStatus  
from task_manager.exceptions import TaskNotFoundError  
from task_manager.logger import create_logger  
  
# Extract the filename without extension  
filename = os.path.splitext(os.path.basename(__file__))[0]  
  
logger = create_logger(logger_name=filename)  
  
  
class DB:  
    def __init__(self):  
        # Create DB Engine  
        self.engine = create_engine("sqlite:///database.db", echo=True)  
        SQLModel.metadata.create_all(self.engine)  
  
    def create_task(self, task: Tasks, session: Session) -> None:  
        """  
        Create a task in the database  
  
        Args:  
            task (Tasks): The task to create  
            Session (Session): The database session  
  
        Returns:  
            int: The ID of the created task  
        """  
        # Write to database  
        logger.info("Creating Task in DB")  
        session.add(task)  
        session.commit()  
  
    def read_task(self, task_id: int, session: Session) -> Tasks:  
        """  
        Read a task from the database  
  
        Args:  
            task_id (int): The ID of the task to read  
            session (Session): The database session  
  
        Returns:  
            Tasks: The task  
        """  
        # Read a task from the database  
        logger.info(f"Reading Task with ID {task_id} from DB")  
        statement = select(Tasks).where(Tasks.id == task_id)  
        result: Tasks = session.exec(statement).first()  
        if result:  
            return result  
        else:  
            logger.error("Task not found")  
            raise TaskNotFoundError(f"Task with ID {task_id} not found")  
  
    def read_tasks(self, session: Session) -> List[Tasks]:  
        """  
        Read all tasks from the database  
  
        Args:  
            session (Session): The database session  
  
        Returns:  
            List[Tasks]: A list of all tasks  
        """  
        # Read all tasks from the database  
        logger.info("Reading all Tasks from DB")  
        statement = select(Tasks)  
        results = session.exec(statement)  
        return [r for r in results]  
  
    def update_task( self,  
        session: Session,  
        task_id: str,  
        task_title: str = None,  
        task_description: str = None,  
        task_status: TaskStatus = None, ) -> None:  
        """  
        Update a task in the database  
  
        Args:  
  
            session (Session): The database session  
            task_id (str): The ID of the task to update  
            task_title (str, optional): The title of the task. Defaults to None.  
            task_description (str, optional): The description of the task. Defaults to None.  
            task_status (TaskStatus, optional): The status of the task. Defaults to None.  
  
        Returns:  
  
            None  
        """  
        # Update a task in the database  
        logger.info(f"Updating Task with ID {task_id} in DB")  
  
        # Get the task  
        statement = select(Tasks).where(Tasks.id == task_id)  
        result = session.exec(statement).first()  
  
        if result:  
            # Update the task  
            if task_title:  
                result.title = task_title  
  
            if task_description:  
                result.description = task_description  
  
            if task_status:  
                result.status = task_status  
  
            updated_at = datetime.now(timezone.utc)  
            result.updated_at = updated_at  
  
            session.add(result)  
            session.commit()  
            logger.info(f"Updated Task with ID {task_id} in DB")  
        else:  
            logger.error(f"Task with ID {task_id} not found")  
            raise TaskNotFoundError(f"Task with ID {task_id} not found")  
  
    def delete_task(self, session: Session, task_id: int) -> None:  
        """  
        Delete a task from the database  
  
        Args:  
            session (Session): The database session  
            task_id (int): The ID of the task to delete  
  
        Returns:  
            None  
        """  
        # Delete a task from the database  
        logger.info("Deleting Task with ID {task_id} from DB")  
        statement = select(Tasks).where(Tasks.id == task_id)  
        results = session.exec(statement)  
        task = results.first()  
  
        if task:  
            # Delete the task  
            session.delete(task)  
            session.commit()  
  
            # Confirm the deletion  
            results_post_delete = session.exec(statement)  
            task_post_delete = results_post_delete.first()  
  
            if task_post_delete is None:  
                logger.info(f"Task with ID {task_id} was confirmed deleted")  
        else:  
            logger.error(f"Task with ID {task_id} not found")  
            raise TaskNotFoundError(f"Task with ID {task_id} not found")  
  
    def delete_all_tasks(self, session: Session) -> None:  
        """  
        Delete all tasks from the database  
  
        Args:  
            session (Session): The database session  
  
        Returns:  
            None  
        """  
        # Delete all tasks from the database  
        logger.info("Deleting all Tasks from DB")  
        statement = select(Tasks)  
        results = session.exec(statement)  
  
        for task in results:  
            session.delete(task)  
            session.commit()  
  
        # Confirm the deletion  
        results_post_delete = session.exec(statement)  
        tasks_post_delete = results_post_delete.all()  
  
        if tasks_post_delete == []:  
            logger.info("All Tasks were confirmed deleted")  
        else:  
            logger.error("All Tasks were not deleted")  
            raise Exception("All Tasks were not deleted")

Don’t be overwhelmed by the size of the source code, it’s quite simple.

Let’s break it down

We have a class DB with an __init__ function that creates the database in SQLite and table for us.
We have individual methods for — create_task , read_task , read_tasks , update_task , delete_task and delete_all_tasks .
Each method takes a Session object while the Engine object is shared across the class. This allows you to bundle several transactions into 1 session and then close it.
There are also checks for basic error handling like if a task_id doesn’t exist.

We also have the following helper files.

task_manager/exceptions.py

"""  
Custom Exceptions for the Task Manager  
"""  
  
  
class TaskNotFoundError(Exception):  
    pass

task_manager/logger.py (allows you to use a custom logger per file)

import logging  
  
  
def create_logger(logger_name: str = __name__):  
    """  
    Create a logger for the module  
  
    Args:  
        logger_name: The name of the logger  
  
    Returns:  
        The logger  
    """  
    logger = logging.getLogger(logger_name)  
    logger.setLevel(logging.INFO)  # Log messages at INFO level and above  
    console_handler = logging.StreamHandler()  
    formatter = logging.Formatter(  
        "%(asctime)s - %(name)s - %(levelname)s - %(message)s"  
    )  
    console_handler.setFormatter(formatter)  
    logger.addHandler(console_handler)  
    return logger

Testing Database Operations

Ideally, you would have developed this code using the popular practice of TDD (Test-Driven Development) — whereby you would write a basic method and then write tests to validate a bunch of success or failure scenarios, tweaking your method in the bargain.

But let’s say we have the code or at least an idea of what we want our db.py to look like.

What functionalities and operations should you test?

Should you test for exceptions? What about the test coverage %?

First, it helps to make a list

What To Test?

Evaluating the functionality we’ve decided that as a bare minimum, we need to cover,

Core CRUD Operations

create task
read task
read all tasks
update task (check that the fields are updated correctly)
update task (check that the updated_at timestamp is updated correctly)
delete task
delete all tasks

Exceptions and Errors

read task — task_id not found
delete task — task_id not found
invalid task model
DB Session failure
DB connection issues

This list is by no means comprehensive and only touches the surface of the type of testing you should consider and implement when working with databases.

In another article, I’ll dive deeper into testing strategy, but here’s an overview of the different types of software testing.

How To Test DB Operations

Let’s now dive into the how, after all knowing what to test is not enough.

First, let’s set up a few fixtures. If you’re not aware of Pytest fixtures, this guide is a great read.

I also strongly urge you to familiarise yourself with Pytest fixture setup and teardown and Pytest fixture scopes.

I’ll discuss why fixtures shortly in a minute.

Define Required Fixtures

tests/conftest.py

import pytest  
from sqlmodel import Session  
from task_manager.db import DB  
from task_manager.model import Tasks, TaskStatus  
  
  
@pytest.fixture  
def task1():  
    """  
    Create a Task 1  
    """  
    task1 = Tasks(  
        title="Go to the Gym",  
        description="Visit Gym at 09:00",  
        status=TaskStatus.NOT_STARTED,  
    )  
    yield task1  
  
  
@pytest.fixture  
def task2():  
    """  
    Create a Task 2  
    """  
    task2 = Tasks(  
        title="Buy Groceries",  
        description="Large shopping list - buy at 12:00",  
        status=TaskStatus.NOT_STARTED,  
    )  
    yield task2  
  
  
@pytest.fixture  
def db_instance(scope="session"):  
    """  
    Create a DB Instance  
    """  
    db = DB()  
    yield db  
  
  
@pytest.fixture  
def session(db_instance, scope="session"):  
    """  
    Create a Session, close after test session, uses `db_instance` fixture  
    """  
    session = Session(db_instance.engine)  
    yield session  
    session.close()  
  
  
@pytest.fixture  
def db_instance_empty(db_instance, session, scope="function"):  
    """  
    Create an Empty DB Instance, uses `db_instance` and `session` fixtures  
    """  
    # Clear DB before test function  
    db_instance.delete_all_tasks(session=session)  
    yield db_instance  
  
    # Clear DB after test function  
    db_instance.delete_all_tasks(session=session)

In this conftest.py file, we’ve defined a few fixtures

task1 (Initialized model of a task)
task2 (initialized model of another task)
db_instance (Class DB instance with session scope)
session (session object with session fixture scope) - use the same session for the whole test session
db_instance_empty (database instance with truncated rows, function fixture scope) - start each test function with an empty database

Now you may ask — why use a different fixture for session and what’s with the db_instance_empty fixture? Why is the db_instance fixture not enough?

Well, the answer lies in test isolation.

You want each of your tests to run on a clean database and avoid any possible cross-contamination. It’s one of the holy grail practices of unit testing.

This fixture allows us to have a clean slate at the start and end of each unit test, assuring you that there is no data leakage.

Note the use of Fixture Setup and Teardown extensively using the yield parameter.

Test DB CRUD Operations

If you haven’t figured it out, CRUD stands for Create, Read, Update and Delete. It’s common terminology in software, database and API development.

The below code tests your core CRUD operations which is the bread and butter of your application.

tests/test_db_core_crud.py

import pytest  
from task_manager.model import TaskStatus  
from task_manager.exceptions import TaskNotFoundError  
  
  
def test_create_and_read_task(db_instance_empty, session, task1):  
    """  
    Test the creation and reading of a task  
    """  
    # Write Task to DB  
    db_instance_empty.create_task(task=task1, session=session)  
  
    # # Read Task from DB  
    task = db_instance_empty.read_task(task_id=1, session=session)  
    assert task.title == task1.title  
    assert task.description == task1.description  
    assert task.status == task1.status  
  
  
def test_read_all_tasks(db_instance_empty, session, task1, task2):  
    """  
    Test the reading of all tasks  
    """  
    # Write 2 Tasks to DB  
    db_instance_empty.create_task(task=task1, session=session)  
    db_instance_empty.create_task(task=task2, session=session)  
  
    # Read all Tasks from DB  
    tasks = db_instance_empty.read_tasks(session=session)  
    assert len(tasks) == 2  
    assert tasks[0].title == task1.title  
    assert tasks[1].title == task2.title  
  
  
def test_read_all_tasks_empty(db_instance_empty, session):  
    """  
    Test the reading of all tasks when the DB is empty  
    """  
    # Read all Tasks from DB  
    tasks = db_instance_empty.read_tasks(session=session)  
    assert len(tasks) == 0  
  
  
def test_delete_task(db_instance_empty, session, task1, task2):  
    """  
    Test the deletion of a task  
    """  
    # Write 2 Tasks to DB  
    db_instance_empty.create_task(task=task1, session=session)  
    db_instance_empty.create_task(task=task2, session=session)  
  
    # Delete Task  
    db_instance_empty.delete_task(session=session, task_id=1)  
  
    # Read Task from DB  
    with pytest.raises(TaskNotFoundError):  
        db_instance_empty.read_task(task_id=1, session=session)  
  
  
def test_delete_all_tasks(db_instance_empty, session, task1, task2):  
    """  
    Test the deletion of all tasks  
    """  
    # Write 2 Tasks to DB  
    db_instance_empty.create_task(task=task1, session=session)  
    db_instance_empty.create_task(task=task2, session=session)  
  
    # Delete all Tasks from DB  
    db_instance_empty.delete_all_tasks(session=session)  
  
    # Read all Tasks from DB  
    tasks = db_instance_empty.read_tasks(session=session)  
    assert len(tasks) == 0  
  
  
def test_update_task(db_instance_empty, session, task1):  
    """  
    Test the updating of a task (status)  
    """  
    # Write Task to DB  
    db_instance_empty.create_task(task=task1, session=session)  
  
    # Update Task  
    db_instance_empty.update_task(  
        session=session,  
        task_id=1,  
        task_status=TaskStatus.COMPLETED,  
        task_title="Wash The Car",  # Change from "Go to the Gym" to "Wash The Car"  
    )  
  
    # Read Task from DB  
    task = db_instance_empty.read_task(task_id=1, session=session)  
  
    # Check Task Status and Updated At  
    assert task.status == TaskStatus.COMPLETED  
    assert task.title == "Wash The Car"  
    assert task.updated_at > task.created_at  
  
    # Check that the description has not changed  
    assert task.description == task1.description  
  
  
def test_update_task_updated_at(db_instance_empty, session, task1):  
    """  
    Test the updating of a task (updated_at)  
    """  
    # Write Task to DB  
    db_instance_empty.create_task(task=task1, session=session)  
  
    # Update Task  
    db_instance_empty.update_task(  
        session=session,  
        task_id=1,  
        task_title="New Title",  
    )  
  
    # Read Task from DB  
    task = db_instance_empty.read_task(task_id=1, session=session)  
  
    # Check Task Updated At is greater than Created At  
    assert task.updated_at > task.created_at

It’s important you read back what you’ve written to the DB, only then you can say with some % of confidence that it works as expected.

Now what about when things don’t go as expected?

Test DB Exceptions

How do you go about testing exceptions?

In our source code above, we defined a custom TaskNotFoundError exception.

What if you try to update or delete a task that’s not in the database?

How about you try to insert a wrong datatype? Are you sure you’re showing the correct messages to your end user?

tests/test_db_exceptions.py

import pytest  
import sqlalchemy  
from task_manager.model import TaskStatus  
from task_manager.exceptions import TaskNotFoundError  
  
  
def test_read_task_not_found(db_instance_empty, session, task1, task2):  
    """  
    Test the reading of a task that does not exist  
    """  
    # Write 2 Tasks to DB  
    db_instance_empty.create_task(task=task1, session=session)  
    db_instance_empty.create_task(task=task2, session=session)  
  
    # Search for Task with ID 100  
    with pytest.raises(TaskNotFoundError):  
        db_instance_empty.read_task(task_id=100, session=session)  
  
  
def test_update_task_not_found(db_instance_empty, session, task1, task2):  
    """  
    Test the updating of a task that does not exist  
    """  
    # Write 2 Tasks to DB  
    db_instance_empty.create_task(task=task1, session=session)  
    db_instance_empty.create_task(task=task2, session=session)  
  
    # Update Task with ID 100  
    with pytest.raises(TaskNotFoundError):  
        db_instance_empty.update_task(  
            session=session,  
            task_id=100,  
            task_status=TaskStatus.COMPLETED,  
        )  
  
  
def test_delete_task_not_found(db_instance_empty, session, task1, task2):  
    """  
    Test the deletion of a task that does not exist  
    """  
    # Write 2 Tasks to DB  
    db_instance_empty.create_task(task=task1, session=session)  
    db_instance_empty.create_task(task=task2, session=session)  
  
    # Delete Task with ID 100  
    with pytest.raises(TaskNotFoundError):  
        db_instance_empty.delete_task(session=session, task_id=100)  
  
  
def test_write_invalid_task(db_instance, session, task1):  
    """  
    Test the Tasks Model with invalid data  
    """  
    # Update task1 with Invalid Data  
    task1.id = "test"  
    task1.status = "test"  
  
    # Write Task to DB and expect an IntegrityError  
    with pytest.raises(  
        sqlalchemy.exc.IntegrityError  
    ):  # Incorrect Data Types for ID and Status raises IntegrityError  
        db_instance.create_task(task=task1, session=session)

Here we have a few tests that simulate the exceptions.

You try to read, update and delete a task whose ID does not exist.

Lastly, you attempt to insert type String values into id and status even though you’ve clearly defined them as int and enum of type TaskStatus respectively.

Note that in the last test case, we’ve used the db_instance fixture instead of the db_instance_empty fixture to avoid trying to delete rows after we raise an sqlalchemy.exe.IntegrityError .

If you attempt to run Delete Operations on an empty database after it raises the integrity error you’ll be faced with a Rollback Error.

Last but not least, have you handled cases where you simply can’t connect to the DB?

Test DB Engine and Connectivity

tests/test_db_engine.py

import pytest  
from sqlalchemy.exc import TimeoutError, OperationalError  
from task_manager.db import DB  
  
  
def test_db_connection_timeout(mocker):  
    # Simulate TimeoutError  
    mocker.patch("task_manager.db.create_engine", side_effect=TimeoutError)  
  
    # Test DB Connection  
    with pytest.raises(TimeoutError):  
        db = DB()  #  
  
  
def test_session_failure(session, mocker, task1):  
    db = DB()  
  
    # Correctly instantiate an OperationalError for mocking  
    operational_error = OperationalError(  
        statement="some SQL statement",  
        params={},  
        orig=Exception("Session Cannot Be Created"),  
    )  
  
    # Mock the `commit` method of the session to raise the mocked OperationalError  
    mocker.patch.object(session, "commit", side_effect=operational_error)  
  
    with pytest.raises(OperationalError):  
        db.create_task(task=task1, session=session)

Simple but often overlooked for their sexier functionality test cousins, testing your DB engine, session and connectivity is an important part.

Connection timeouts are common especially when working with databases in a cloud or on-prem VPC.

Now how should you test this?

Should you set up a database in a private VPC and try to access it via the Internet? No, that’s not necessary for unit testing.

But you can implement a test database in an integration or end-to-end testing scenarios which you’d typically run on a CI/CD release platform. More on that later.

Simple answer, Mock external services and systems.

With 1 simple line, you can tell Pytest what this Class, Function or Variable should return.

In our case, we said, task_manager.db.create_engine should return a TimeoutError.

In the second test, session.commit should return an SQLAlchemy OperationalError.

And then we just test it using pytest.raises.

Testing exception handling is a very important topic in unit testing as most things don’t go as planned and you have to be prepared for the worst.

Moreover, if you’re working with user input, external APIs and systems and so on.

Testing Results

Now to run these tests

You can just run pytest from your terminal.

$ pytest

pytest-db-testing-results

You can also choose to classify tests based on Pytest Markers which offer a convenient way to selectively run tests using the -m flag e.g. pytest -m unit will run all tests marked with a @pytest.mark.unit marker.

Note — Database Sessions

A small note on database sessions.

If you read through the source code above, you’ll see that we’ve initialized the engine parameter only once and session parameter passed to every method.

Why is this? Why not use a unique session per operation and close it?

Well then you wouldn’t easily be able to read after you’ve written and would end up with a DetachedInstanceError .

It’s way more efficient to have the option to bundle several SQL operations into 1 session which allows you more flexibility.

You can choose to use 1 session per operation or 1 session for the whole unit test, which in a front-end sense can also be then tied to browser sessions.

Ultimately it’s up to you, but bear in mind to always test the functionality of the source code, not the implementation.

Tiangolo talks more about sessions and engines in the SQLModel official docs.

Best Practices in Database Testing Unit Testing

Use an ORM Tool — Can’t stress enough how much of a safety net this gives you compared to traditional SQL execution and the risk of SQL Injection.
Start with Empty DB, use Fixtures — Using an empty database per test greatly reduces the risk of data contamination and leakage across tests, affecting results.
Seed the DB if necessary — If you need a full database to test with, don’t hesitate to “seed” the database with test data, just make sure you return to the same database state after each test.
Run tests locally with in-memory DB like SQLite and then run tests against a dedicated Test DB instance on the CI/CD platforms.
Create test data automatically using tools like Hypothesis or Pytest Parametrization.
Use Mock data to simulate real-world scenarios. This reduces dependencies and ensures test isolation.
Check Data Integrity and Constraints — Check that the data inserted is actually what you expect and that DB constraints are followed e.g. Unique Primary Keys or Foreign Keys.
Document Your Tests: Keep your tests well-documented. This helps other developers understand the purpose and implementation of each test, making maintenance easier. Also document # TODO tests.
Regularly Review and Refactor Tests: Regularly review and refactor your tests to ensure they remain relevant and effective.

In-Memory DB vs External DB

This age-old question, should you use an in-memory DB like SQLite or TinyDB or a traditional one like Postgres or SQL Server?

My take — if you’re testing simple unit test operations and need tests to run fast, an in-memory DB should suffice.

But if you’re testing DB-specific functions, routines or procedures, you should connect to the real one.

Also in your CI/CD Pipelines, you may have integration tests to test the “real integration” in that case use a real test database.

Drawbacks of Testing with the Real DB

So what are the drawbacks of using a real database?

Maybe you don’t have access to it?

This means you have to ask the IT or Platform team to set up the database or grant you network and security access, which could take a while depending on your organization.

Database queries also add significant latency depending on the network and how close you are to it.

Lastly, an important point mentioned above is the database state.

It’s so important to test against a fixed initial state and this might be tough or even impossible if others are writing to the database at the same time.

Drawbacks of In-Memory DBs

I personally love SQLite and TinyDB, however I’m also aware of their limitations.

Can’t test DB-specific procedures and functions.
In-memory DB data is volatile which means that data can be lost if the process is terminated or if the system crashes. While some in-memory databases offer persistence options, they are primarily designed for temporary storage.
In-memory databases can’t be scaled to high volumes that simulate real-world operations.
It’s also tough to test concurrency and locking which is a common challenge when working with multi-user environments.
Security features like access controls, encryption, and auditing may be less robust than those in traditional databases.
In-memory databases may not fully support complex relationships and queries as efficiently as traditional relational databases.

Test API Layer — Flask, FastAPI, Django

As an application developer, you’d rarely expose code that directly interfaces with the database.

Why? Because this opens up doors for all kinds of possible issues.

You don’t want your users to accidentally corrupt data through loopholes in your backend system.

That’s where the API layer comes in.

API frameworks like FastAPI (which I’m a big fan of), Django or Flask bridge the gap here and offer a clean secure layer of user interactions with the database.

They allow you to define endpoints, response codes, and payloads and take care of security and other bits and pieces.

FastAPI is built with Pydantic in mind, thereby allowing to securely validate requests and responses, ensuring database integrity and security.

Other applications add an extra step like a CLI layer if needed.

Test frameworks also come with their Test Clients which you can neatly define as fixtures and leverage some inbuilt mock methods.

Conclusion

This has been a fairly long article.

I tried to cover as much as I could but at over 4,000 words my logical head told me to wrap this up.

In this article, you learned so much about database unit testing.

You started with the importance of ORM tools over traditional SQL queries, and then learned how to build a simple task manager application that can read and write a Tasks model to an in-memory database.

All of this using modern tooling like SQLModel, SQLAlchemy and Pydantic.

You learned how to define your test strategy and efficiently use fixtures with the correct scope to handle database setup and teardown, ensuring DB state and test isolation.

Last you learned about DB Sessions, database unit testing best practices and drawbacks of in-memory vs traditional DBs.

With all of this experience, you’re now fully equipped to build your backend application and handle database queries in a highly efficient and pro manner.

Don’t forget to keep building on your unit tests, starting with the most obvious and grow to fix bugs you inevitably discover in production.

Happy testing 🚀

If you have ideas for improvement or like me to cover anything specific, please send me a message via Twitter, GitHub or Email.

Till the next time… Cheers!

Additional Reading

Link To Example Code Used In This Article
How Pytest Fixtures Can Help You Write More Readable And Efficient Tests
Automated Python Unit Testing Made Easy with Pytest and GitHub Actions
Regression Tests
What is Setup and Teardown in Pytest? (Importance of a Clean Test Environment)
What Are Pytest Fixture Scopes? (How To Choose The Best Scope For Your Test)
Python Unit Testing Best Practices For Building Reliable Applications
Introduction to Pytest Mocking - What It Is and Why You Need It
How To Test Python Exception Handling Using Pytest Assert (A Simple Guide)
How to Use Hypothesis and Pytest for Robust Property-Based Testing in Python
How to Effortlessly Generate Unit Test Cases with Pytest Parameterized Tests
Pydantic - Official Docs
SQLModel - Official Docs