7.3N+1 Query Detection

The Problem

The N+1 problem occurs when resolving a list of N items triggers N additional database queries instead of one batched query.

Query: users(first: 10) { friends { name } }

N+1 pattern (BAD):
  Query 1: SELECT * FROM users LIMIT 10     -- 1 query
  Query 2: SELECT * FROM friends WHERE user_id = 1  -- N queries
  Query 3: SELECT * FROM friends WHERE user_id = 2
  ...
  Query 11: SELECT * FROM friends WHERE user_id = 10
  Total: 11 queries

Batched pattern (GOOD):
  Query 1: SELECT * FROM users LIMIT 10
  Query 2: SELECT * FROM friends WHERE user_id IN (1,2,3,...,10)
  Total: 2 queries

Detecting N+1 in Tests

class TestGraphQLN1Detection:
    """Detect N+1 query patterns in GraphQL resolvers."""

    def test_user_friends_not_n_plus_1(self, graphql_client, query_logger):
        """Fetching users with friends should batch, not N+1."""
        query = """
        query {
          users(first: 10) {
            id
            name
            friends {
              id
              name
            }
          }
        }
        """
        query_logger.reset()
        response = graphql_client.execute(query)

        # Count database queries that were executed
        db_queries = query_logger.get_queries()
        user_count = len(response["data"]["users"])

        # N+1 would be: 1 (list users) + N (fetch friends per user)
        # Correct batching: 1 (list users) + 1 (batch friends)
        assert len(db_queries) <= 3, (
            f"Expected <= 3 DB queries (batched), got {len(db_queries)} "
            f"for {user_count} users. Likely N+1 pattern:\n"
            + "\n".join(q.sql for q in db_queries)
        )

    def test_nested_resolution_performance(self, graphql_client):
        """Complex nested query should complete within SLA."""
        query = """
        query {
          products(first: 20) {
            id
            name
            category { name }
            reviews(first: 5) {
              rating
              author { name }
            }
          }
        }
        """
        import time
        start = time.monotonic()
        response = graphql_client.execute(query)
        duration = time.monotonic() - start

        assert duration < 2.0, (
            f"Nested query took {duration:.2f}s, SLA is 2.0s. "
            f"Possible N+1 in reviews or author resolution."
        )

    def test_dataloader_batching(self, graphql_client, query_logger):
        """Verify DataLoader batches related entity lookups."""
        query = """
        query {
          orders(first: 20) {
            id
            customer {
              id
              name
            }
            items {
              product {
                id
                name
              }
            }
          }
        }
        """
        query_logger.reset()
        response = graphql_client.execute(query)
        db_queries = query_logger.get_queries()

        # With DataLoader: ~4 queries (orders, customers, items, products)
        # Without DataLoader: 1 + 20 + 20*items + ... queries
        assert len(db_queries) <= 6, (
            f"Expected <= 6 DB queries with DataLoader, got {len(db_queries)}. "
            f"Verify DataLoader is configured for customer and product resolvers."
        )

Setting Up Query Logging

# conftest.py
import logging

class QueryLogger:
    """Capture database queries for N+1 detection."""

    def __init__(self):
        self.queries = []
        self._handler = None

    def start(self):
        """Start capturing SQL queries."""
        self._handler = logging.getLogger('sqlalchemy.engine').addHandler(
            self._capture_handler()
        )

    def reset(self):
        self.queries = []

    def get_queries(self):
        return self.queries

    def _capture_handler(self):
        class CaptureHandler(logging.Handler):
            def __init__(self, logger):
                super().__init__()
                self.logger = logger

            def emit(self, record):
                if "SELECT" in record.getMessage():
                    self.logger.queries.append(
                        CapturedQuery(sql=record.getMessage(), timestamp=time.time())
                    )
        return CaptureHandler(self)

@pytest.fixture
def query_logger():
    logger = QueryLogger()
    logger.start()
    return logger