c.table if isinstance(c, expression.ColumnClause)
else c for c in select._for_update_arg.of)
tmp += " OF " + ", ".join(
- self.process(table, ashint=True, **kw)
+ self.process(table, ashint=True, use_schema=False, **kw)
for table in tables
)
return text
def visit_table(self, table, asfrom=False, iscrud=False, ashint=False,
- fromhints=None, **kwargs):
+ fromhints=None, use_schema=True, **kwargs):
if asfrom or ashint:
- if getattr(table, "schema", None):
+ if use_schema and getattr(table, "schema", None):
ret = self.preparer.quote_schema(table.schema) + \
"." + self.preparer.quote(table.name)
else:
"WHERE mytable_1.myid = %(myid_1)s FOR UPDATE OF mytable_1"
)
+ def test_for_update_with_schema(self):
+ m = MetaData()
+ table1 = Table(
+ 'mytable', m,
+ Column('myid'),
+ Column('name'),
+ schema='testschema'
+ )
+
+ self.assert_compile(
+ table1.select(table1.c.myid == 7).with_for_update(of=table1),
+ "SELECT testschema.mytable.myid, testschema.mytable.name "
+ "FROM testschema.mytable "
+ "WHERE testschema.mytable.myid = %(myid_1)s "
+ "FOR UPDATE OF mytable")
+
def test_reserved_words(self):
table = Table("pg_table", MetaData(),
Column("col1", Integer),