Nplusone

1.0.0 · active · verified Tue Apr 14

Nplusone is a Python library designed to detect N+1 query problems in Object-Relational Mappers (ORMs) during development. It supports popular ORMs like SQLAlchemy, Peewee, and the Django ORM. The library monitors database interactions and emits warnings or raises exceptions when potentially inefficient lazy loads or unnecessary eager loads are detected. The current version is 1.0.0, released in May 2018, and it appears to have a low release cadence, indicating a mature and stable project.

Warnings

Install

Imports

Quickstart

This quickstart demonstrates how to use `nplusone` with SQLAlchemy. It sets up a simple ORM model, adds some data, and then uses the `Profiler` context manager to detect N+1 queries when accessing related `Artist` objects within a loop without eager loading. Warnings will be logged to the console.

import logging
from nplusone.core.profiler import Profiler
import nplusone.ext.sqlalchemy

# Configure a logger to capture nplusone warnings
logger = logging.getLogger('nplusone')
logger.setLevel(logging.WARN)
handler = logging.StreamHandler()
logger.addHandler(handler)

# --- Simulate an SQLAlchemy setup ---
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Artist(Base):
    __tablename__ = 'artists'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    songs = relationship('Song', back_populates='artist')

class Song(Base):
    __tablename__ = 'songs'
    id = Column(Integer, primary_key=True)
    title = Column(String)
    artist_id = Column(Integer, ForeignKey('artists.id'))
    artist = relationship('Artist', back_populates='songs')

engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()

# Add some data
artist1 = Artist(name='Artist One')
artist2 = Artist(name='Artist Two')
session.add_all([artist1, artist2])
session.commit()

song1 = Song(title='Song A', artist=artist1)
song2 = Song(title='Song B', artist=artist1)
song3 = Song(title='Song C', artist=artist2)
session.add_all([song1, song2, song3])
session.commit()

# --- Nplusone profiling ---
print('--- Starting nplusone profiling ---')
with Profiler():
    songs = session.query(Song).all()
    print(f'Fetched {len(songs)} songs.')
    # This will trigger N+1 queries if artists are not eagerly loaded
    for song in songs:
        print(f'  Song: {song.title}, Artist: {song.artist.name}')

print('--- Profiling complete ---')

# Example of how to raise an NPlusOneError for tests (optional)
# from nplusone.core.exceptions import NPlusOneError
# import os
# os.environ['NPLUSONE_RAISE'] = 'True' # Set this environment variable or config option
# try:
#     with Profiler():
#         songs = session.query(Song).all()
#         for song in songs:
#             _ = song.artist.name
# except NPlusOneError as e:
#     print(f'Caught expected NPlusOneError: {e}')
# os.environ['NPLUSONE_RAISE'] = '' # Reset for other tests

view raw JSON →