How does one go about testing queries in SQLAlchemy? For example suppose we have this models.py
from sqlalchemy import ( Column, Integer, String, ) from sqlalchemy.ext.declarative import declarative_base Base = declarative_base() class Panel(Base): __tablename__ = 'Panels' id = Column(Integer, primary_key=True) category = Column(Integer, nullable=False) platform = Column(String, nullable=False) region = Column(String, nullable=False) def __init__(self, category, platform, region): self.category = category self.platform = platform self.region = region def __repr__(self): return ( "<Panel('{self.category}', '{self.platform}', " "'{self.region}')>".format(self=self) )
and this tests.py
import unittest from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from models import Base, Panel class TestQuery(unittest.TestCase): engine = create_engine('sqlite:///:memory:') Session = sessionmaker(bind=engine) session = Session() def setUp(self): Base.metadata.create_all(self.engine) self.session.add(Panel(1, 'ion torrent', 'start')) self.session.commit() def tearDown(self): Base.metadata.drop_all(self.engine) def test_query_panel(self): expected = [Panel(1, 'ion torrent', 'start')] result = self.session.query(Panel).all() self.assertEqual(result, expected)
When we try running the test, it fails, even though the two Panels look identical.
$ nosetests F ====================================================================== FAIL: test_query_panel (tests.TestQuery) ---------------------------------------------------------------------- Traceback (most recent call last): File "/Users/clasher/tmp/tests.py", line 31, in test_query_panel self.assertEqual(result, expected) AssertionError: Lists differ: [<Panel('1', 'ion torrent', 's... != [<Panel('1', 'ion torrent', 's... First differing element 0: <Panel('1', 'ion torrent', 'start')> <Panel('1', 'ion torrent', 'start')> [<Panel('1', 'ion torrent', 'start')>, <Panel('2', 'ion torrent', 'end')>] ---------------------------------------------------------------------- Ran 1 test in 0.063s FAILED (failures=1)
One solution I've found is to make a query for every single instance I expect to find in the query:
class TestQuery(unittest.TestCase): ... def test_query_panel(self): expected = [ (1, 'ion torrent', 'start'), (2, 'ion torrent', 'end') ] successful = True # Check to make sure every expected item is in the query try: for category, platform, region in expected: self.session.query(Panel).filter_by( category=category, platform=platform, region=region).one() except (NoResultFound, MultipleResultsFound): successful = False self.assertTrue(successful) # Check to make sure no unexpected items are in the query self.assertEqual(self.session.query(Panel).count(), len(expected))
This strikes me as pretty ugly, though, and I'm not even getting to the point where I have a complex filtered query that I'm trying to test. Is there a more elegant solution, or do I always have to manually make a bunch of individual queries?
All SELECT statements generated by SQLAlchemy ORM are constructed by Query object. It provides a generative interface, hence successive calls return a new Query object, a copy of the former with additional criteria and options associated with it.
The grouping is done with the group_by() query method, which takes the column to use for the grouping as an argument, same as the GROUP BY counterpart in SQL. The statement ends by calling subquery() , which tells SQLAlchemy that our intention for this query is to use it inside a bigger query instead of on its own.
SQLAlchemy is the ORM of choice for working with relational databases in python. The reason why SQLAlchemy is so popular is because it is very simple to implement, helps you develop your code quicker and doesn't require knowledge of SQL to get started.
method sqlalchemy.orm.Query. all() Return the results represented by this Query as a list. This results in an execution of the underlying SQL statement. The Query object, when asked to return either a sequence or iterator that consists of full ORM-mapped entities, will deduplicate entries based on primary key.
your original test is on the right track, you just have to do one of two things: either make sure that two Panel
objects of the same primary key identity compare as True
:
class Panel(Base): # ... def __eq__(self, other): return isinstance(other, Panel) and other.id == self.id
or you can organize your test such that you make sure you're checking against the same Panel
instance (because here we take advantage of the identity map):
class TestQuery(unittest.TestCase): def setUp(self): self.engine = create_engine('sqlite:///:memory:') self.session = Session(engine) Base.metadata.create_all(self.engine) self.panel = Panel(1, 'ion torrent', 'start') self.session.add(self.panel) self.session.commit() def tearDown(self): Base.metadata.drop_all(self.engine) def test_query_panel(self): expected = [self.panel] result = self.session.query(Panel).all() self.assertEqual(result, expected)
as far as the engine/session setup/teardown, I'd go for a pattern where you use a single engine for all tests, and assuming your schema is fixed, a single schema for all tests, then you make sure the data you work with is performed within a transaction that can be rolled back. The Session
can be made to work this way, such that calling commit()
doesn't actually commit the "real" transaction, by wrapping the whole test within an explicit Transaction
. The example at https://docs.sqlalchemy.org/en/latest/orm/session_transaction.html#joining-a-session-into-an-external-transaction-such-as-for-test-suites illustrates this usage. Having a ":memory:" engine on every test fixture will take up a lot of memory and not really scale out to other databases besides SQLite.
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With