Oleg Broytmann wrote:
On Thu, Apr 13, 2006 at 05:12:20PM -0500, Charles Duffy wrote:
It would probably have been a good thing if this patch actually fixed
the issue it purported to, yes?
Can you also add a test?
The attached version breaks fewer tests; fixes the ones which it breaks;
and adds a few new test cases.
I've also validated that test_subqueries.test_5perform_direct fails with
the unpatched SQLObject build, such that this is in fact a valid test.
def test_5perform_direct():
insert()
select = TestIn1.select(TestIn1.q.col1 == Select(TestIn2.q.col2,
where=(TestIn2.q.col2 == "test")))
assert select.count() == 1
Index: sqlobject/dbconnection.py
===================================================================
--- sqlobject/dbconnection.py (revision 1695)
+++ sqlobject/dbconnection.py (working copy)
@@ -604,9 +604,9 @@
# in the SQLObject class.
def _SO_update(self, so, values):
- self.query("UPDATE %s SET %s WHERE %s = %s" %
+ self.query("UPDATE %s SET %s WHERE %s = (%s)" %
(so.sqlmeta.table,
- ", ".join(["%s = %s" % (dbName, self.sqlrepr(value))
+ ", ".join(["%s = (%s)" % (dbName, self.sqlrepr(value))
for dbName, value in values]),
so.sqlmeta.idName,
self.sqlrepr(so.id)))
@@ -615,14 +615,14 @@
columns = ", ".join(columnNames)
if columns:
return self.queryOne(
- "SELECT %s FROM %s WHERE %s = %s" %
+ "SELECT %s FROM %s WHERE %s = (%s)" %
(columns,
so.sqlmeta.table,
so.sqlmeta.idName,
self.sqlrepr(so.id)))
else:
return self.queryOne(
- "SELECT NULL FROM %s WHERE %s = %s" %
+ "SELECT NULL FROM %s WHERE %s = (%s)" %
(so.sqlmeta.table,
so.sqlmeta.idName,
self.sqlrepr(so.id)))
@@ -635,7 +635,7 @@
raise ValueError, "'column' and 'value' tuples must be of the same size"
columns = []
for i in xrange(len(column)):
- columns.append("%s = %s" % (column[i], self.sqlrepr(value[i])))
+ columns.append("(%s) = (%s)" % (column[i], self.sqlrepr(value[i])))
condition = ' AND '.join(columns)
return self.queryOne("SELECT %s FROM %s WHERE %s" %
(", ".join(columnNames),
@@ -643,20 +643,20 @@
condition))
def _SO_delete(self, so):
- self.query("DELETE FROM %s WHERE %s = %s" %
+ self.query("DELETE FROM %s WHERE %s = (%s)" %
(so.sqlmeta.table,
so.sqlmeta.idName,
self.sqlrepr(so.id)))
def _SO_selectJoin(self, soClass, column, value):
- return self.queryAll("SELECT %s FROM %s WHERE %s = %s" %
+ return self.queryAll("SELECT %s FROM %s WHERE %s = (%s)" %
(soClass.sqlmeta.idName,
soClass.sqlmeta.table,
column,
self.sqlrepr(value)))
def _SO_intermediateJoin(self, table, getColumn, joinColumn, value):
- return self.queryAll("SELECT %s FROM %s WHERE %s = %s" %
+ return self.queryAll("SELECT %s FROM %s WHERE %s = (%s)" %
(getColumn,
table,
joinColumn,
@@ -664,7 +664,7 @@
def _SO_intermediateDelete(self, table, firstColumn, firstValue,
secondColumn, secondValue):
- self.query("DELETE FROM %s WHERE %s = %s AND %s = %s" %
+ self.query("DELETE FROM %s WHERE %s = (%s) AND %s = (%s)" %
(table,
firstColumn,
self.sqlrepr(firstValue),
Index: sqlobject/sqlbuilder.py
===================================================================
--- sqlobject/sqlbuilder.py (revision 1695)
+++ sqlobject/sqlbuilder.py (working copy)
@@ -230,7 +230,13 @@
self.expr1 = expr1
self.expr2 = expr2
def __sqlrepr__(self, db):
- return "(%s %s %s)" % (sqlrepr(self.expr1, db), self.op, sqlrepr(self.expr2, db))
+ s1 = sqlrepr(self.expr1, db)
+ s2 = sqlrepr(self.expr2, db)
+ if s1[0] != '(' and s1 != 'NULL':
+ s1 = '(' + s1 + ')'
+ if s2[0] != '(' and s2 != 'NULL':
+ s2 = '(' + s2 + ')'
+ return "(%s %s %s)" % (s1, self.op, s2)
def components(self):
return [self.expr1, self.expr2]
def execute(self, executor):
@@ -884,7 +890,7 @@
self.expr = expr
self.string = string
def __sqlrepr__(self, db):
- return "(%s %s %s)" % (sqlrepr(self.expr, db), self.op, sqlrepr(self.string, db))
+ return "(%s %s (%s))" % (sqlrepr(self.expr, db), self.op, sqlrepr(self.string, db))
def components(self):
return [self.expr, self.string]
def execute(self, executor):
@@ -912,7 +918,7 @@
else:
return "LIKE"
def __sqlrepr__(self, db):
- return "(%s %s %s)" % (
+ return "(%s %s (%s))" % (
sqlrepr(self.expr, db), self._get_op(db), sqlrepr(self.string, db)
)
def execute(self, executor):
Index: sqlobject/tests/test_joins_conditional.py
===================================================================
--- sqlobject/tests/test_joins_conditional.py (revision 1695)
+++ sqlobject/tests/test_joins_conditional.py (working copy)
@@ -35,7 +35,7 @@
on_condition=(TestJoin1.q.col1 == TestJoin2.q.col2))
)
assert str(select) == \
- "SELECT test_join1.id, test_join1.col1 FROM test_join1 LEFT JOIN test_join2 ON (test_join1.col1 = test_join2.col2) WHERE 1 = 1"
+ "SELECT test_join1.id, test_join1.col1 FROM test_join1 LEFT JOIN test_join2 ON ((test_join1.col1) = (test_join2.col2)) WHERE 1 = 1"
def test_3perform_join():
setup()
Index: sqlobject/tests/test_subqueries.py
===================================================================
--- sqlobject/tests/test_subqueries.py (revision 1695)
+++ sqlobject/tests/test_subqueries.py (working copy)
@@ -40,9 +40,20 @@
setup()
select = TestIn1.select(NOTEXISTS(Select(TestIn2.q.col2, where=(Outer(TestIn1).q.col1 == TestIn2.q.col2))))
assert str(select) == \
- "SELECT test_in1.id, test_in1.col1 FROM test_in1 WHERE NOT EXISTS (SELECT test_in2.col2 FROM test_in2 WHERE (test_in1.col1 = test_in2.col2))"
+ "SELECT test_in1.id, test_in1.col1 FROM test_in1 WHERE NOT EXISTS (SELECT test_in2.col2 FROM test_in2 WHERE ((test_in1.col1) = (test_in2.col2)))"
def test_4perform_exists():
insert()
select = TestIn1.select(EXISTS(Select(TestIn2.q.col2, where=(Outer(TestIn1).q.col1 == TestIn2.q.col2))))
assert len(list(select)) == 2
+
+def test_4syntax_direct():
+ setup()
+ select = TestIn1.select(TestIn1.q.col1 == Select(TestIn2.q.col2, where=(TestIn2.q.col2 == "test")))
+ assert str(select) == \
+ "SELECT test_in1.id, test_in1.col1 FROM test_in1 WHERE ((test_in1.col1) = (SELECT test_in2.col2 FROM test_in2 WHERE ((test_in2.col2) = ('test'))))"
+
+def test_4perform_direct():
+ insert()
+ select = TestIn1.select(TestIn1.q.col1 == Select(TestIn2.q.col2, where=(TestIn2.q.col2 == "test")))
+ assert select.count() == 1
Index: sqlobject/tests/test_converters.py
===================================================================
--- sqlobject/tests/test_converters.py (revision 1695)
+++ sqlobject/tests/test_converters.py (working copy)
@@ -163,7 +163,7 @@
def test_op():
instance = SQLOp('and', 'this', 'that')
- assert sqlrepr(instance, 'mysql') == "('this' AND 'that')"
+ assert sqlrepr(instance, 'mysql') == "(('this') AND ('that'))"
def test_call():
instance = SQLCall('test', ('test',))
Index: sqlobject/tests/test_aliases.py
===================================================================
--- sqlobject/tests/test_aliases.py (revision 1695)
+++ sqlobject/tests/test_aliases.py (working copy)
@@ -15,7 +15,7 @@
alias = Alias(JoinAlias)
select = JoinAlias.select(JoinAlias.q.parent == alias.q.name)
assert str(select) == \
- "SELECT join_alias.id, join_alias.name, join_alias.parent FROM join_alias AS join_alias_alias1, join_alias WHERE (join_alias.parent = join_alias_alias1.name)"
+ "SELECT join_alias.id, join_alias.name, join_alias.parent FROM join_alias AS join_alias_alias1, join_alias WHERE ((join_alias.parent) = (join_alias_alias1.name))"
def test_2perform_join():
setupClass(JoinAlias)
Index: sqlobject/main.py
===================================================================
--- sqlobject/main.py (revision 1695)
+++ sqlobject/main.py (working copy)
@@ -1483,7 +1483,7 @@
if col.cascade == False:
# Found a restriction
restrict = True
- query.append("%s = %s" % (col.dbName, self.id))
+ query.append("%s = (%s)" % (col.dbName, self.id))
if col.cascade == 'null':
setnull = col.name
elif col.cascade:
