from sqlalchemy import testing
from sqlalchemy.testing import fixtures, AssertsCompiledSQL
from sqlalchemy import Integer, String, MetaData, Table, Column, select, \
- func, not_, cast, text, tuple_, exists, delete, update, bindparam,\
+ func, not_, cast, text, tuple_, exists, update, bindparam,\
insert, literal, and_, null, type_coerce, alias, or_, literal_column,\
Float, TIMESTAMP, Numeric, Date, Text, collate, union, except_,\
intersect, union_all, Boolean, distinct, join, outerjoin, asc, desc,\
"AND myothertable.othername = mytable_1.name",
dialect=mssql.dialect())
- def test_delete(self):
- self.assert_compile(
- delete(table1, table1.c.myid == 7),
- "DELETE FROM mytable WHERE mytable.myid = :myid_1")
- self.assert_compile(
- table1.delete().where(table1.c.myid == 7),
- "DELETE FROM mytable WHERE mytable.myid = :myid_1")
- self.assert_compile(
- table1.delete().where(table1.c.myid == 7).\
- where(table1.c.name == 'somename'),
- "DELETE FROM mytable WHERE mytable.myid = :myid_1 "
- "AND mytable.name = :name_1")
-
- def test_delete_prefix(self):
- stmt = table1.delete().prefix_with("A", "B", dialect="mysql").\
- prefix_with("C", "D")
- self.assert_compile(stmt,
- "DELETE A B C D FROM mytable",
- dialect=mysql.dialect()
- )
- self.assert_compile(stmt,
- "DELETE C D FROM mytable")
-
- def test_aliased_delete(self):
- talias1 = table1.alias('t1')
- self.assert_compile(
- delete(talias1).where(talias1.c.myid == 7),
- "DELETE FROM mytable AS t1 WHERE t1.myid = :myid_1")
-
- def test_correlated_delete(self):
- # test a non-correlated WHERE clause
- s = select([table2.c.othername], table2.c.otherid == 7)
- u = delete(table1, table1.c.name == s)
- self.assert_compile(u, "DELETE FROM mytable WHERE mytable.name = "
- "(SELECT myothertable.othername FROM myothertable "
- "WHERE myothertable.otherid = :otherid_1)")
-
- # test one that is actually correlated...
- s = select([table2.c.othername], table2.c.otherid == table1.c.myid)
- u = table1.delete(table1.c.name == s)
- self.assert_compile(u,
- "DELETE FROM mytable WHERE mytable.name = (SELECT "
- "myothertable.othername FROM myothertable WHERE "
- "myothertable.otherid = mytable.myid)")
-
def test_binds_that_match_columns(self):
"""test bind params named after column names
replace the normal SET/VALUES generation."""
--- /dev/null
+#! coding:utf-8
+
+from sqlalchemy import Column, Integer, String, Table, delete, select
+from sqlalchemy.dialects import mysql
+from sqlalchemy.testing import AssertsCompiledSQL, fixtures
+
+
+class _DeleteTestBase(object):
+ @classmethod
+ def define_tables(cls, metadata):
+ Table('mytable', metadata,
+ Column('myid', Integer),
+ Column('name', String),
+ Column('description', String))
+ Table('myothertable', metadata,
+ Column('otherid', Integer),
+ Column('othername', String))
+
+
+class DeleteTest(_DeleteTestBase, fixtures.TablesTest, AssertsCompiledSQL):
+ __dialect__ = 'default'
+
+ def test_delete(self):
+ table1 = self.tables.mytable
+
+ self.assert_compile(
+ delete(table1, table1.c.myid == 7),
+ 'DELETE FROM mytable WHERE mytable.myid = :myid_1')
+
+ self.assert_compile(
+ table1.delete().where(table1.c.myid == 7),
+ 'DELETE FROM mytable WHERE mytable.myid = :myid_1')
+
+ self.assert_compile(
+ table1.delete().
+ where(table1.c.myid == 7).
+ where(table1.c.name == 'somename'),
+ 'DELETE FROM mytable '
+ 'WHERE mytable.myid = :myid_1 '
+ 'AND mytable.name = :name_1')
+
+ def test_prefix_with(self):
+ table1 = self.tables.mytable
+
+ stmt = table1.delete().\
+ prefix_with('A', 'B', dialect='mysql').\
+ prefix_with('C', 'D')
+
+ self.assert_compile(stmt,
+ 'DELETE C D FROM mytable')
+
+ self.assert_compile(stmt,
+ 'DELETE A B C D FROM mytable',
+ dialect=mysql.dialect())
+
+ def test_alias(self):
+ table1 = self.tables.mytable
+
+ talias1 = table1.alias('t1')
+ stmt = delete(talias1).where(talias1.c.myid == 7)
+
+ self.assert_compile(stmt,
+ 'DELETE FROM mytable AS t1 WHERE t1.myid = :myid_1')
+
+ def test_correlated(self):
+ table1, table2 = self.tables.mytable, self.tables.myothertable
+
+ # test a non-correlated WHERE clause
+ s = select([table2.c.othername], table2.c.otherid == 7)
+ self.assert_compile(delete(table1, table1.c.name == s),
+ 'DELETE FROM mytable '
+ 'WHERE mytable.name = ('
+ 'SELECT myothertable.othername '
+ 'FROM myothertable '
+ 'WHERE myothertable.otherid = :otherid_1'
+ ')')
+
+ # test one that is actually correlated...
+ s = select([table2.c.othername], table2.c.otherid == table1.c.myid)
+ self.assert_compile(table1.delete(table1.c.name == s),
+ 'DELETE FROM mytable '
+ 'WHERE mytable.name = ('
+ 'SELECT myothertable.othername '
+ 'FROM myothertable '
+ 'WHERE myothertable.otherid = mytable.myid'
+ ')')