--- /dev/null
+.. change::
+ :tags: bug, orm, regression
+ :tickets: 6414
+
+ Fixed regression where :meth:`_orm.Query.filter_by` would not work if the
+ lead entity were a SQL function or other expression derived from the
+ primary entity in question, rather than a simple entity or column of that
+ entity. Additionally, improved the behavior of
+ :meth:`_sql.Select.filter_by` overall to work with column expressions even
+ in a non-ORM context.
import re
from . import roles
+from . import visitors
from .traversals import HasCacheKey # noqa
from .traversals import HasCopyInternals # noqa
from .traversals import MemoizedHasCacheKey # noqa
return bind
+def _entity_namespace(entity):
+ """Return the nearest .entity_namespace for the given entity.
+
+ If not immediately available, does an iterate to find a sub-element
+ that has one, if any.
+
+ """
+ try:
+ return entity.entity_namespace
+ except AttributeError:
+ for elem in visitors.iterate(entity):
+ if hasattr(elem, "entity_namespace"):
+ return elem.entity_namespace
+ else:
+ raise
+
+
def _entity_namespace_key(entity, key):
"""Return an entry from an entity_namespace.
"""
- ns = entity.entity_namespace
try:
+ ns = _entity_namespace(entity)
return getattr(ns, key)
except AttributeError as err:
util.raise_(
f = f._is_clone_of
return s
+ @property
+ def entity_namespace(self):
+ raise AttributeError(
+ "This SQL expression has no entity namespace "
+ "with which to filter from."
+ )
+
def __getstate__(self):
d = self.__dict__.copy()
d.pop("_is_clone_of", None)
# expect the columns of tables and subqueries to be leaf nodes.
return []
+ @property
+ def entity_namespace(self):
+ if self.table is not None:
+ return self.table.entity_namespace
+ else:
+ return super(ColumnClause, self).entity_namespace
+
@HasMemoized.memoized_attribute
def _from_objects(self):
t = self.table
from . import schema
from . import sqltypes
from . import util as sqlutil
+from .base import _entity_namespace
from .base import ColumnCollection
from .base import Executable
from .base import Generative
else:
return super(FunctionElement, self).self_group(against=against)
+ @property
+ def entity_namespace(self):
+ """overrides FromClause.entity_namespace as functions are generally
+ column expressions and not FromClauses.
+
+ """
+ # ideally functions would not be fromclauses but we failed to make
+ # this adjustment in 1.4
+ return _entity_namespace(self.clause_expr)
+
class FunctionAsBinary(BinaryExpression):
_traverse_internals = [
checkparams={"email_address_1": "ed@ed.com", "name_1": "ed"},
)
+ def test_filter_by_against_function(self):
+ """test #6414
+
+ this is related to #6401 where the fact that Function is a
+ FromClause, an architectural mistake that we unfortunately did not
+ fix, is confusing the use of entity_namespace etc.
+
+ """
+ User = self.classes.User
+ sess = fixture_session()
+
+ q1 = sess.query(func.count(User.id)).filter_by(name="ed")
+
+ self.assert_compile(
+ q1,
+ "SELECT count(users.id) AS count_1 FROM users "
+ "WHERE users.name = :name_1",
+ )
+
+ def test_filter_by_against_cast(self):
+ """test #6414"""
+ User = self.classes.User
+ sess = fixture_session()
+
+ q1 = sess.query(cast(User.id, Integer)).filter_by(name="ed")
+
+ self.assert_compile(
+ q1,
+ "SELECT CAST(users.id AS INTEGER) AS users_id FROM users "
+ "WHERE users.name = :name_1",
+ )
+
+ def test_filter_by_against_binary(self):
+ """test #6414"""
+ User = self.classes.User
+ sess = fixture_session()
+
+ q1 = sess.query(User.id == 5).filter_by(name="ed")
+
+ self.assert_compile(
+ q1,
+ "SELECT users.id = :id_1 AS anon_1 FROM users "
+ "WHERE users.name = :name_1",
+ )
+
+ def test_filter_by_against_label(self):
+ """test #6414"""
+ User = self.classes.User
+ sess = fixture_session()
+
+ q1 = sess.query(User.id.label("foo")).filter_by(name="ed")
+
+ self.assert_compile(
+ q1,
+ "SELECT users.id AS foo FROM users " "WHERE users.name = :name_1",
+ )
+
def test_empty_filters(self):
User = self.classes.User
sess = fixture_session()
+from sqlalchemy import cast
from sqlalchemy import Column
from sqlalchemy import exc
from sqlalchemy import ForeignKey
+from sqlalchemy import func
from sqlalchemy import Integer
from sqlalchemy import MetaData
from sqlalchemy import select
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import fixtures
+
table1 = table(
"mytable",
column("myid", Integer),
checkparams={"data_1": "p1", "data_2": "c1", "otherid_1": 5},
)
- def test_filter_by_no_property(self):
+ def test_filter_by_from_col(self):
+ stmt = select(table1.c.myid).filter_by(name="foo")
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid FROM mytable WHERE mytable.name = :name_1",
+ )
+
+ def test_filter_by_from_func(self):
+ """test #6414"""
+ stmt = select(func.count(table1.c.myid)).filter_by(name="foo")
+ self.assert_compile(
+ stmt,
+ "SELECT count(mytable.myid) AS count_1 "
+ "FROM mytable WHERE mytable.name = :name_1",
+ )
+
+ def test_filter_by_from_func_not_the_first_arg(self):
+ """test #6414"""
+ stmt = select(func.bar(True, table1.c.myid)).filter_by(name="foo")
+ self.assert_compile(
+ stmt,
+ "SELECT bar(:bar_2, mytable.myid) AS bar_1 "
+ "FROM mytable WHERE mytable.name = :name_1",
+ )
+
+ def test_filter_by_from_cast(self):
+ """test #6414"""
+ stmt = select(cast(table1.c.myid, Integer)).filter_by(name="foo")
+ self.assert_compile(
+ stmt,
+ "SELECT CAST(mytable.myid AS INTEGER) AS myid "
+ "FROM mytable WHERE mytable.name = :name_1",
+ )
+
+ def test_filter_by_from_binary(self):
+ """test #6414"""
+ stmt = select(table1.c.myid == 5).filter_by(name="foo")
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid = :myid_1 AS anon_1 "
+ "FROM mytable WHERE mytable.name = :name_1",
+ )
+
+ def test_filter_by_from_label(self):
+ """test #6414"""
+ stmt = select(table1.c.myid.label("some_id")).filter_by(name="foo")
+ self.assert_compile(
+ stmt,
+ "SELECT mytable.myid AS some_id "
+ "FROM mytable WHERE mytable.name = :name_1",
+ )
+
+ def test_filter_by_no_property_from_table(self):
assert_raises_message(
exc.InvalidRequestError,
'Entity namespace for "mytable" has no property "foo"',
foo="bar",
)
+ def test_filter_by_no_property_from_col(self):
+ assert_raises_message(
+ exc.InvalidRequestError,
+ 'Entity namespace for "mytable.myid" has no property "foo"',
+ select(table1.c.myid).filter_by,
+ foo="bar",
+ )
+
def test_select_tuple_outer(self):
stmt = select(tuple_(table1.c.myid, table1.c.name))