warn if query.get() used with existing criterion

diff --git a/lib/sqlalchemy/orm/query.py b/lib/sqlalchemy/orm/query.py
index eb0d49147c0ecb7d9fdfa173e6f8de494213d93c..20f18eecbc8510f8a1561eef622f53b2407613e7 100644 (file)
--- a/lib/sqlalchemy/orm/query.py
+++ b/lib/sqlalchemy/orm/query.py
@@ -10,6 +10,7 @@ from sqlalchemy.sql import expression, visitors, operators
 from sqlalchemy.orm import mapper, object_mapper
 from sqlalchemy.orm import util as mapperutil
 from itertools import chain
+import warnings
 
 __all__ = ['Query', 'QueryContext']
 
@@ -55,6 +56,10 @@ class Query(object):
     
     def _no_criterion(self):
         q = self._clone()
+        
+        if q._criterion or q._statement or q._from_obj != [self.table]:
+            warnings.warn(RuntimeWarning("Query.get() being called on a Query with existing criterion; criterion is being ignored."))
+            
         q._from_obj = [self.table]
         q._alias_ids = {}
         q._joinpoint = self.mapper

diff --git a/test/orm/query.py b/test/orm/query.py
index 158576f1766edce0ed5f0af0c4c6ce5dba9c99aa..4e5f0485589ed641637d281f37e6bd22004e699b 100644 (file)
--- a/test/orm/query.py
+++ b/test/orm/query.py
@@ -63,8 +63,19 @@ class GetTest(QueryTest):
     
     def test_no_criterion(self):
         """test that get()/load() does not use preexisting filter/etc. criterion"""
-        
+
         s = create_session()
+
+        import warnings
+        warnings.filterwarnings("error", r".*Query.*")
+        print "---------------------------------------"
+        try:
+            s.query(User).join('addresses').filter(Address.user_id==8).get(7)
+            assert False
+        except RuntimeWarning, e:
+            assert str(e) == "Query.get() being called on a Query with existing criterion; criterion is being ignored."
+
+        warnings.filterwarnings("once", r".*Query.*")
         
         assert s.query(User).filter(User.id==7).get(19) is None
 
@@ -86,6 +97,8 @@ class GetTest(QueryTest):
         assert s.query(User).join('addresses').filter(Address.user_id==8).get(7).id == u.id
         
         assert s.query(User).join('addresses').filter(Address.user_id==8).load(7).id == u.id
+
+
         
     def test_unique_param_names(self):
         class SomeUser(object):