How to Master SQLAlchemy I/O: Testing Queries in CI to Prevent Database Disasters 🚨

Felipe Cardoso Martins — Wed, 08 Apr 2026 18:41:56 +0000

It’s 3:00 AM. Your pager is screaming.

The application is completely unresponsive, the database CPU is pegged at 100%, and connection pools are exhausted. Desperate customers with critical systems offline are flooding the support channels. To stop the bleeding, your team scales up to the biggest AWS RDS instance available, literally burning thousands of USD per minute just to keep the lights on.

You scramble to find the root cause, expecting a massive infrastructure failure. Instead, you find a single, seemingly harmless Python loop that was recently deployed.

Your CI pipeline was completely green. All the unit tests passed. The API returned the correct JSON schema. But beneath that green checkmark, your ORM was quietly executing 5,000 individual SELECT statements per request.

Testing what your application does is no longer enough. If you aren't testing how it communicates with your database, you are exposing your business to catastrophic financial and operational risk. Let's explore how to take control of your execution footprint.

🏢 The Cultural Divide: Whose Problem is the Database?

For years, software development has suffered from a toxic, siloed mentality: "Writing the code is my job; the database performance is the DBA's problem."

This culture is a massive financial liability. C-Level executives are painfully aware that the "database black box" directly inflates cloud infrastructure bills. You cannot simply throw more expensive AWS compute power at poorly optimized I/O.

At the same time, developers are constantly pushed to deliver features faster, relying heavily on Object-Relational Mappers (ORMs) to abstract away the SQL layer. But abstractions are not magic. Building a resilient engineering culture requires developers to take absolute ownership of their execution footprint. You must understand the exact cost of the code you write.

🛡️ Engineering Excellence Disclaimer

Let's get one thing straight: SQLAlchemy is not slow. When a database reaches a critical state, it is almost never the fault of the ORM itself. The ORM is doing exactly what you commanded it to do.

Modern engineering demands "agnostic generalist specialists." You do not need to be a DBA, but you must understand relational mechanics and make architectural decisions about your I/O layer:

The Python GC & Object Hydration Trap: SQLAlchemy does far more than just translate Python to SQL. It manages an IdentityMap, tracks the "dirty state" of every record, and hydrates complex Python objects. If you lazily load 10,000 rows as full ORM models instead of lightweight tuples, you aren't just stressing the database—you are suffocating Python's memory. When the Garbage Collector (GC) eventually kicks in to clean up thousands of discarded objects, your application's CPU will spike and the event loop will stall. You must know when to yield raw tuples or use load_only.
The JOIN Illusion: It is a common misconception that a massive JOIN is always the best way to avoid an N+1 problem. While a JOIN utilizes database indexes efficiently, it can easily destroy your networking performance. If you join a root table with a heavily populated child table, the database sends the root data duplicated across every single row over the network. This Cartesian explosion causes terrible I/O bottlenecks.
The Two-Query Strategy: Often, it is vastly superior to execute a first query, aggregate the IDs in memory, and then execute a second query using an IN (...) clause. This completely eliminates the N+1 problem while keeping the network payload incredibly lean.
Virtual Tables and Pushdown Logic: When dealing with heavy aggregations, doing the math in Python memory is a critical mistake. It is almost always better to create a virtual table (like a View or a CTE) to push the computational weight down to the database engine, returning only the final, lightweight result to your application.

You must be in control of these decisions. pytest-capquery exists to make this invisible I/O battle visual. It puts you in control, commander.

💡 The Solution: Bridging the Gap with `pytest-capquery`

We need a way to incentivize developers to care about database I/O without forcing them to manually write and maintain brittle, hardcoded SQL assertions in their test suites.

This is why pytest-capquery was created. It intercepts the SQLAlchemy engine at the driver level, providing a strict, chronological timeline of your application's execution footprint.

For the Business: This is about protecting the bottom line. Catching a database regression in CI preserves your system's SLA and safeguards your customer reputation. You avoid emergency weekend patches, furious customers with offline security panels, and the sheer financial drain of desperately scaling up your cloud infrastructure just to keep the platform breathing.
For Developers: It uses a zero-friction snapshot workflow. You don't write SQL strings; the test suite generates them for you. If an N+1 regression occurs, the test fails immediately. You use the snapshot as a debugging mechanism to continuously improve your query logic.
For DBAs: It automatically generates physical .sql files. DBAs can review these raw SQL artifacts during Pull Requests to validate query plans and indexes without ever reading a line of Python code.

🛠️ Getting Started: Proving Your Execution Footprint

Let's look at how to protect a critical domain—like monitoring Alarm Panels and their associated Sensors—using a real PostgreSQL integration database.

1. The Setup (`conftest.py`)

First, we provision a tangible PostgreSQL engine to ensure our tests replicate production-grade execution topologies. We configure the postgres_capquery fixture to intercept the engine.

from typing import Generator
import pytest
from sqlalchemy import create_engine, Engine, text
from sqlalchemy.orm import Session, sessionmaker
from pytest_capquery.plugin import CapQueryWrapper
from pytest_capquery.snapshot import SnapshotManager
from tests.models import Base

@pytest.fixture(scope="session")
def postgres_engine() -> Generator[Engine, None, None]:
    engine = create_engine("postgresql+psycopg2://postgres@localhost:5432/capquery_test")
    Base.metadata.create_all(engine)
    yield engine
    Base.metadata.drop_all(engine)
    engine.dispose()

@pytest.fixture(scope="function")
def postgres_session(postgres_engine: Engine) -> Generator[Session, None, None]:
    SessionMaker = sessionmaker(bind=postgres_engine)
    session = SessionMaker()
    session.execute(text("TRUNCATE TABLE alarm_panels, sensors RESTART IDENTITY CASCADE"))
    session.commit()
    yield session
    session.rollback()
    session.close()

@pytest.fixture(scope="function")
def postgres_capquery(
    postgres_engine: Engine, capquery_context: SnapshotManager
) -> Generator[CapQueryWrapper, None, None]:
    with CapQueryWrapper(postgres_engine, snapshot_manager=capquery_context) as captured:
        yield captured

2. The Test (`test_snapshot.py`)

Instead of guessing how many queries are executed, we wrap our business logic in the capture(assert_snapshot=True) context manager.

import pytest
from sqlalchemy.orm import joinedload
from tests.models import AlarmPanel, Sensor

pytestmark = pytest.mark.xdist_group("e2e_postgres")

def test_insert_and_select_snapshot(postgres_session, postgres_capquery):
    with postgres_capquery.capture(assert_snapshot=True):
        panel = AlarmPanel(mac_address="00:11:22:33:44:55", is_online=True)
        sensor = Sensor(name="Front Door", sensor_type="Contact")
        panel.sensors.append(sensor)

        postgres_session.add(panel)
        postgres_session.flush()

        queried_panel = (
            postgres_session.query(AlarmPanel)
            .options(joinedload(AlarmPanel.sensors))
            .filter_by(mac_address="00:11:22:33:44:55")
            .first()
        )
        assert queried_panel is not None

3. The Universal Artifact (`.sql` Snapshot)

When you run your test suite, pytest-capquery generates this exact file. This is the ultimate source of truth. If a developer accidentally alters the fetching strategy and destroys your networking performance, the test will instantly fail because the query structure and count will deviate from this approved baseline.

-- CAPQUERY: Query 1
-- EXPECTED_PARAMS: None
-- PHASE: 1
BEGIN

-- CAPQUERY: Query 2
-- EXPECTED_PARAMS: {'mac_address': '00:11:22:33:44:55', 'is_online': True}
-- PHASE: 1
INSERT INTO alarm_panels (mac_address, is_online)
VALUES (%(mac_address)s, %(is_online)s) RETURNING alarm_panels.id

-- CAPQUERY: Query 3
-- EXPECTED_PARAMS: {'panel_id': 1, 'name': 'Front Door', 'sensor_type': 'Contact'}
-- PHASE: 1
INSERT INTO sensors (panel_id, name, sensor_type)
VALUES (%(panel_id)s, %(name)s, %(sensor_type)s) RETURNING sensors.id

-- CAPQUERY: Query 4
-- EXPECTED_PARAMS: {'mac_address_1': '00:11:22:33:44:55', 'param_1': 1}
-- PHASE: 1
SELECT anon_1.alarm_panels_id AS anon_1_alarm_panels_id,
       anon_1.alarm_panels_mac_address AS anon_1_alarm_panels_mac_address,
       anon_1.alarm_panels_is_online AS anon_1_alarm_panels_is_online,
       sensors_1.id AS sensors_1_id,
       sensors_1.panel_id AS sensors_1_panel_id,
       sensors_1.name AS sensors_1_name,
       sensors_1.sensor_type AS sensors_1_sensor_type
FROM
  (SELECT alarm_panels.id AS alarm_panels_id,
          alarm_panels.mac_address AS alarm_panels_mac_address,
          alarm_panels.is_online AS alarm_panels_is_online
   FROM alarm_panels
   WHERE alarm_panels.mac_address = %(mac_address_1)s
   LIMIT %(param_1)s) AS anon_1
LEFT OUTER JOIN sensors AS sensors_1 ON anon_1.alarm_panels_id = sensors_1.panel_id

🚀 Stop Guessing, Start Asserting

The database is the beating heart of your application. Leaving its performance up to chance and ORM black boxes is no longer an option.

By integrating tools like pytest-capquery into your CI pipeline, you transform performance testing from an afterthought into a rigorous, automated standard. You protect your cloud budget, you give your DBAs the transparency they desperately need, and you empower yourself to truly command the systems you build.

Stop guessing your execution footprint. Profile your test suite today:

🔗 fmartins/pytest-capquery on GitHub

pip install pytest-capquery

Together we can do more! If you care about engineering excellence and robust testing, jump into the repository. Issues, discussions, and Pull Requests are always welcome. Let's build a culture that respects the database.

🛑 Stop Testing Your Code and Ignoring Your Database (Catching N+1 in Pytest)

Felipe Cardoso Martins — Wed, 01 Apr 2026 23:18:45 +0000

Your green CI pipeline might be lying to you. 🚨

It tells you the code works, but it’s quietly hiding the N+1 database disaster that will bring down your production environment next week.

As Python & SQLAlchemy developers, we spend hours writing tests to assert our application’s final state, but we treat the database layer like a complete black box. We test what the application does, but completely ignore how it does it.

The business cost of this abstraction is expensive. 💸
Every inefficient query and silent lazy-load that slips into the main branch directly inflates your cloud bill and degrades the user experience.

I got tired of this, so I built and open-sourced pytest-capquery. 🛠️

🎯 What it does

pytest-capquery treats SQL queries as first-class citizens in your Pytest suite. By intercepting the SQLAlchemy engine at the driver level, it enforces a strict, chronological timeline of your execution footprint.

Instead of just checking if a function returns True, you can rigorously assert deterministic I/O. If an N+1 regression slips in, the build fails instantly. 💥

🐛 The N+1 Problem in Action

Let's say a developer forgets to use joinedload on a simple query:

def test_demonstrate_n_plus_one_problem(db_session, capquery):
    capquery.statements.clear()

    panels = db_session.query(AlarmPanel).all()

    for panel in panels:
        _ = panel.sensors

    capquery.assert_executed_queries(
        "SELECT ... FROM alarm_panels",
        ("SELECT ... FROM sensors WHERE ? = sensors.panel_id", (1,)),
        ("SELECT ... FROM sensors WHERE ? = sensors.panel_id", (2,)),
        ("SELECT ... FROM sensors WHERE ? = sensors.panel_id", (3,))
    )

If someone drops the joinedload optimization, pytest-capquery exposes the exact lazy-loading queries.

✅ The Fix

When you optimize the query, your test ensures the database behaves exactly as intended:

from sqlalchemy.orm import joinedload

def test_avoid_n_plus_one_queries(db_session, capquery):
    capquery.statements.clear()

    panels = db_session.query(AlarmPanel).options(joinedload(AlarmPanel.sensors)).all()

    for panel in panels:
        _ = panel.sensors

    capquery.assert_executed_queries(
        """
        SELECT ...
        FROM alarm_panels
        LEFT OUTER JOIN sensors AS sensors_1 ON alarm_panels.id = sensors_1.panel_id
        """
    )

Stop blaming the ORM for performance bottlenecks and start profiling your tests! 📈 Lock down your database performance, drastically increase your software resilience, and stop merging regressions.

👇 Check out the project and let me know what you think:

🔗 Repository: fmartins/pytest-capquery
📦 Install: pip install pytest-capquery

Forem: Felipe Cardoso Martins