mirror of
https://github.com/sqlalchemy/sqlalchemy.git
synced 2026-05-16 13:47:20 -04:00
Mike Bayer
7e864fc7b1
As the test suite has widespread use of many patterns that are deprecated, enable SQLALCHEMY_WARN_20 globally for the test suite but then break the warnings filter out into a whole list of all the individual warnings we are looking for. this way individual changesets can target a specific class of warning, as many of these warnings will indivdidually affect dozens of files and potentially hundreds of lines of code. Many warnings are also resolved here as this patch started out that way. From this point forward there should be changesets that target a subset of the warnings at a time. For expediency, updates some migration 2.0 docs for ORM as well. Change-Id: I98b8defdf7c37b818b3824d02f7668e3f5f31c94
384 lines
12 KiB
Python
384 lines
12 KiB
Python
#! coding:utf-8
|
|
|
|
from sqlalchemy import and_
|
|
from sqlalchemy import delete
|
|
from sqlalchemy import exc
|
|
from sqlalchemy import exists
|
|
from sqlalchemy import ForeignKey
|
|
from sqlalchemy import Integer
|
|
from sqlalchemy import or_
|
|
from sqlalchemy import select
|
|
from sqlalchemy import String
|
|
from sqlalchemy import testing
|
|
from sqlalchemy.dialects import mysql
|
|
from sqlalchemy.engine import default
|
|
from sqlalchemy.testing import assert_raises_message
|
|
from sqlalchemy.testing import AssertsCompiledSQL
|
|
from sqlalchemy.testing import eq_
|
|
from sqlalchemy.testing import expect_deprecated
|
|
from sqlalchemy.testing import fixtures
|
|
from sqlalchemy.testing.schema import Column
|
|
from sqlalchemy.testing.schema import Table
|
|
|
|
|
|
class _DeleteTestBase(object):
|
|
@classmethod
|
|
def define_tables(cls, metadata):
|
|
Table(
|
|
"mytable",
|
|
metadata,
|
|
Column("myid", Integer),
|
|
Column("name", String(30)),
|
|
Column("description", String(50)),
|
|
)
|
|
Table(
|
|
"myothertable",
|
|
metadata,
|
|
Column("otherid", Integer),
|
|
Column("othername", String(30)),
|
|
)
|
|
|
|
|
|
class DeleteTest(_DeleteTestBase, fixtures.TablesTest, AssertsCompiledSQL):
|
|
__dialect__ = "default"
|
|
|
|
def test_delete_literal_binds(self):
|
|
table1 = self.tables.mytable
|
|
|
|
stmt = table1.delete().where(table1.c.name == "jill")
|
|
|
|
self.assert_compile(
|
|
stmt,
|
|
"DELETE FROM mytable WHERE mytable.name = 'jill'",
|
|
literal_binds=True,
|
|
)
|
|
|
|
def test_delete(self):
|
|
table1 = self.tables.mytable
|
|
|
|
self.assert_compile(
|
|
delete(table1).where(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_where_empty(self):
|
|
table1 = self.tables.mytable
|
|
|
|
with expect_deprecated():
|
|
self.assert_compile(
|
|
table1.delete().where(and_()), "DELETE FROM mytable"
|
|
)
|
|
with expect_deprecated():
|
|
self.assert_compile(
|
|
table1.delete().where(or_()), "DELETE FROM mytable"
|
|
)
|
|
|
|
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_non_correlated_select(self):
|
|
table1, table2 = self.tables.mytable, self.tables.myothertable
|
|
|
|
# test a non-correlated WHERE clause
|
|
s = select(table2.c.othername).where(table2.c.otherid == 7)
|
|
self.assert_compile(
|
|
delete(table1).where(table1.c.name == s.scalar_subquery()),
|
|
"DELETE FROM mytable "
|
|
"WHERE mytable.name = ("
|
|
"SELECT myothertable.othername "
|
|
"FROM myothertable "
|
|
"WHERE myothertable.otherid = :otherid_1"
|
|
")",
|
|
)
|
|
|
|
def test_correlated_select(self):
|
|
table1, table2 = self.tables.mytable, self.tables.myothertable
|
|
|
|
# test one that is actually correlated...
|
|
s = select(table2.c.othername).where(table2.c.otherid == table1.c.myid)
|
|
self.assert_compile(
|
|
table1.delete().where(table1.c.name == s.scalar_subquery()),
|
|
"DELETE FROM mytable "
|
|
"WHERE mytable.name = ("
|
|
"SELECT myothertable.othername "
|
|
"FROM myothertable "
|
|
"WHERE myothertable.otherid = mytable.myid"
|
|
")",
|
|
)
|
|
|
|
|
|
class DeleteFromCompileTest(
|
|
_DeleteTestBase, fixtures.TablesTest, AssertsCompiledSQL
|
|
):
|
|
# DELETE FROM is also tested by individual dialects since there is no
|
|
# consistent syntax. here we use the StrSQLcompiler which has a fake
|
|
# syntax.
|
|
|
|
__dialect__ = "default_enhanced"
|
|
|
|
def test_delete_extra_froms(self):
|
|
table1, table2 = self.tables.mytable, self.tables.myothertable
|
|
|
|
stmt = table1.delete().where(table1.c.myid == table2.c.otherid)
|
|
self.assert_compile(
|
|
stmt,
|
|
"DELETE FROM mytable , myothertable "
|
|
"WHERE mytable.myid = myothertable.otherid",
|
|
)
|
|
|
|
def test_correlation_to_extra(self):
|
|
table1, table2 = self.tables.mytable, self.tables.myothertable
|
|
|
|
stmt = (
|
|
table1.delete()
|
|
.where(table1.c.myid == table2.c.otherid)
|
|
.where(
|
|
~exists()
|
|
.where(table2.c.otherid == table1.c.myid)
|
|
.where(table2.c.othername == "x")
|
|
.correlate(table2)
|
|
)
|
|
)
|
|
|
|
self.assert_compile(
|
|
stmt,
|
|
"DELETE FROM mytable , myothertable WHERE mytable.myid = "
|
|
"myothertable.otherid AND NOT (EXISTS "
|
|
"(SELECT * FROM mytable WHERE myothertable.otherid = "
|
|
"mytable.myid AND myothertable.othername = :othername_1))",
|
|
)
|
|
|
|
def test_dont_correlate_to_extra(self):
|
|
table1, table2 = self.tables.mytable, self.tables.myothertable
|
|
|
|
stmt = (
|
|
table1.delete()
|
|
.where(table1.c.myid == table2.c.otherid)
|
|
.where(
|
|
~exists()
|
|
.where(table2.c.otherid == table1.c.myid)
|
|
.where(table2.c.othername == "x")
|
|
.correlate()
|
|
)
|
|
)
|
|
|
|
self.assert_compile(
|
|
stmt,
|
|
"DELETE FROM mytable , myothertable WHERE mytable.myid = "
|
|
"myothertable.otherid AND NOT (EXISTS "
|
|
"(SELECT * FROM myothertable, mytable "
|
|
"WHERE myothertable.otherid = "
|
|
"mytable.myid AND myothertable.othername = :othername_1))",
|
|
)
|
|
|
|
def test_autocorrelate_error(self):
|
|
table1, table2 = self.tables.mytable, self.tables.myothertable
|
|
|
|
stmt = (
|
|
table1.delete()
|
|
.where(table1.c.myid == table2.c.otherid)
|
|
.where(
|
|
~exists()
|
|
.where(table2.c.otherid == table1.c.myid)
|
|
.where(table2.c.othername == "x")
|
|
)
|
|
)
|
|
|
|
assert_raises_message(
|
|
exc.InvalidRequestError,
|
|
".*returned no FROM clauses due to auto-correlation.*",
|
|
stmt.compile,
|
|
dialect=default.StrCompileDialect(),
|
|
)
|
|
|
|
|
|
class DeleteFromRoundTripTest(fixtures.TablesTest):
|
|
__backend__ = True
|
|
|
|
@classmethod
|
|
def define_tables(cls, metadata):
|
|
Table(
|
|
"mytable",
|
|
metadata,
|
|
Column("myid", Integer),
|
|
Column("name", String(30)),
|
|
Column("description", String(50)),
|
|
)
|
|
Table(
|
|
"myothertable",
|
|
metadata,
|
|
Column("otherid", Integer),
|
|
Column("othername", String(30)),
|
|
)
|
|
Table(
|
|
"users",
|
|
metadata,
|
|
Column(
|
|
"id", Integer, primary_key=True, test_needs_autoincrement=True
|
|
),
|
|
Column("name", String(30), nullable=False),
|
|
)
|
|
Table(
|
|
"addresses",
|
|
metadata,
|
|
Column(
|
|
"id", Integer, primary_key=True, test_needs_autoincrement=True
|
|
),
|
|
Column("user_id", None, ForeignKey("users.id")),
|
|
Column("name", String(30), nullable=False),
|
|
Column("email_address", String(50), nullable=False),
|
|
)
|
|
Table(
|
|
"dingalings",
|
|
metadata,
|
|
Column(
|
|
"id", Integer, primary_key=True, test_needs_autoincrement=True
|
|
),
|
|
Column("address_id", None, ForeignKey("addresses.id")),
|
|
Column("data", String(30)),
|
|
)
|
|
Table(
|
|
"update_w_default",
|
|
metadata,
|
|
Column("id", Integer, primary_key=True),
|
|
Column("x", Integer),
|
|
Column("ycol", Integer, key="y"),
|
|
Column("data", String(30), onupdate=lambda: "hi"),
|
|
)
|
|
|
|
@classmethod
|
|
def fixtures(cls):
|
|
return dict(
|
|
users=(
|
|
("id", "name"),
|
|
(7, "jack"),
|
|
(8, "ed"),
|
|
(9, "fred"),
|
|
(10, "chuck"),
|
|
),
|
|
addresses=(
|
|
("id", "user_id", "name", "email_address"),
|
|
(1, 7, "x", "jack@bean.com"),
|
|
(2, 8, "x", "ed@wood.com"),
|
|
(3, 8, "x", "ed@bettyboop.com"),
|
|
(4, 8, "x", "ed@lala.com"),
|
|
(5, 9, "x", "fred@fred.com"),
|
|
),
|
|
dingalings=(
|
|
("id", "address_id", "data"),
|
|
(1, 2, "ding 1/2"),
|
|
(2, 5, "ding 2/5"),
|
|
),
|
|
)
|
|
|
|
@testing.requires.delete_from
|
|
def test_exec_two_table(self):
|
|
users, addresses = self.tables.users, self.tables.addresses
|
|
dingalings = self.tables.dingalings
|
|
|
|
with testing.db.connect() as conn:
|
|
conn.execute(dingalings.delete()) # fk violation otherwise
|
|
|
|
conn.execute(
|
|
addresses.delete()
|
|
.where(users.c.id == addresses.c.user_id)
|
|
.where(users.c.name == "ed")
|
|
)
|
|
|
|
expected = [
|
|
(1, 7, "x", "jack@bean.com"),
|
|
(5, 9, "x", "fred@fred.com"),
|
|
]
|
|
self._assert_table(addresses, expected)
|
|
|
|
@testing.requires.delete_from
|
|
def test_exec_three_table(self):
|
|
users = self.tables.users
|
|
addresses = self.tables.addresses
|
|
dingalings = self.tables.dingalings
|
|
|
|
testing.db.execute(
|
|
dingalings.delete()
|
|
.where(users.c.id == addresses.c.user_id)
|
|
.where(users.c.name == "ed")
|
|
.where(addresses.c.id == dingalings.c.address_id)
|
|
)
|
|
|
|
expected = [(2, 5, "ding 2/5")]
|
|
self._assert_table(dingalings, expected)
|
|
|
|
@testing.requires.delete_from
|
|
def test_exec_two_table_plus_alias(self):
|
|
users, addresses = self.tables.users, self.tables.addresses
|
|
dingalings = self.tables.dingalings
|
|
|
|
with testing.db.connect() as conn:
|
|
conn.execute(dingalings.delete()) # fk violation otherwise
|
|
a1 = addresses.alias()
|
|
conn.execute(
|
|
addresses.delete()
|
|
.where(users.c.id == addresses.c.user_id)
|
|
.where(users.c.name == "ed")
|
|
.where(a1.c.id == addresses.c.id)
|
|
)
|
|
|
|
expected = [(1, 7, "x", "jack@bean.com"), (5, 9, "x", "fred@fred.com")]
|
|
self._assert_table(addresses, expected)
|
|
|
|
@testing.requires.delete_from
|
|
def test_exec_alias_plus_table(self):
|
|
users, addresses = self.tables.users, self.tables.addresses
|
|
dingalings = self.tables.dingalings
|
|
|
|
d1 = dingalings.alias()
|
|
|
|
testing.db.execute(
|
|
delete(d1)
|
|
.where(users.c.id == addresses.c.user_id)
|
|
.where(users.c.name == "ed")
|
|
.where(addresses.c.id == d1.c.address_id)
|
|
)
|
|
|
|
expected = [(2, 5, "ding 2/5")]
|
|
self._assert_table(dingalings, expected)
|
|
|
|
def _assert_table(self, table, expected):
|
|
stmt = table.select().order_by(table.c.id)
|
|
eq_(testing.db.execute(stmt).fetchall(), expected)
|