--- /dev/null
+.. change::
+ :tags: bug, asyncio
+
+ Fixed issue in semi-private ``await_only()`` and ``await_fallback()``
+ concurrency functions where the given awaitable would remain un-awaited if
+ the function threw a ``GreenletError``, which could cause "was not awaited"
+ warnings later on if the program continued. In this case, the given
+ awaitable is now cancelled before the exception is thrown.
>>> orm_stmt = (
... select(User)
... .from_statement(union_stmt)
- ... .options(with_expression(User.book_count, union_stmt.c.book_count))
+ ... .options(with_expression(User.book_count, union_stmt.selected_columns.book_count))
... )
>>> for user in session.scalars(orm_stmt):
... print(f"Username: {user.name} Number of books: {user.book_count}")
from sqlalchemy import join
from sqlalchemy import select
from sqlalchemy import String
-from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import clear_mappers
from sqlalchemy.orm import column_property
+from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import deferred
from sqlalchemy.orm import exc as orm_exc
def terminate(self):
self._connection.terminate()
+ self._started = False
@staticmethod
def _default_name_func():
def close(self):
try:
self.await_(self._connection.close())
+ except ValueError:
+ # this is undocumented for aiosqlite, that ValueError
+ # was raised if .close() was called more than once, which is
+ # both not customary for DBAPI and is also not a DBAPI.Error
+ # exception. This is now fixed in aiosqlite via my PR
+ # https://github.com/omnilib/aiosqlite/pull/238, so we can be
+ # assured this will not become some other kind of exception,
+ # since it doesn't raise anymore.
+
+ pass
except Exception as error:
self._handle_exception(error)
"The :class:`.AliasOption` object is not necessary "
"for entities to be matched up to a query that is established "
"via :meth:`.Query.from_statement` and now does nothing.",
+ enable_warnings=False, # AliasOption itself warns
)
def contains_alias(alias: Union[Alias, Subquery]) -> AliasOption:
r"""Return a :class:`.MapperOption` that will indicate to the
to the given list of columns
"""
+ return self._values_no_warn(*columns)
+ _values = values
+
+ def _values_no_warn(
+ self, *columns: _ColumnsClauseArgument[Any]
+ ) -> Iterable[Any]:
if not columns:
return iter(())
q = self._clone().enable_eagerloads(False)
q.load_options += {"_yield_per": 10}
return iter(q) # type: ignore
- _values = values
-
@util.deprecated(
"1.4",
":meth:`_query.Query.value` "
"""
try:
- return next(self.values(column))[0] # type: ignore
+ return next(self._values_no_warn(column))[0] # type: ignore
except StopIteration:
return None
@util.deprecated(
"1.4",
- "The :class:`.AliasOption` is not necessary "
+ "The :class:`.AliasOption` object is not necessary "
"for entities to be matched up to a query that is established "
"via :meth:`.Query.from_statement` and now does nothing.",
)
Note that the test suite sets SAWarning warnings to raise exceptions.
""" # noqa
- return _expect_warnings(sa_exc.SAWarning, messages, **kw)
+ return _expect_warnings_sqla_only(sa_exc.SAWarning, messages, **kw)
@contextlib.contextmanager
def expect_deprecated(*messages, **kw):
- return _expect_warnings(sa_exc.SADeprecationWarning, messages, **kw)
+ return _expect_warnings_sqla_only(
+ sa_exc.SADeprecationWarning, messages, **kw
+ )
def expect_deprecated_20(*messages, **kw):
- return _expect_warnings(sa_exc.Base20DeprecationWarning, messages, **kw)
+ return _expect_warnings_sqla_only(
+ sa_exc.Base20DeprecationWarning, messages, **kw
+ )
def emits_warning_on(db, *messages):
_EXC_CLS = None
+def _expect_warnings_sqla_only(
+ exc_cls,
+ messages,
+ regex=True,
+ search_msg=False,
+ assert_=True,
+):
+ """SQLAlchemy internal use only _expect_warnings().
+
+ Alembic is using _expect_warnings() directly, and should be updated
+ to use this new interface.
+
+ """
+ return _expect_warnings(
+ exc_cls,
+ messages,
+ regex=regex,
+ search_msg=search_msg,
+ assert_=assert_,
+ raise_on_any_unexpected=True,
+ )
+
+
@contextlib.contextmanager
def _expect_warnings(
exc_cls,
raise_on_any_unexpected=False,
squelch_other_warnings=False,
):
-
global _FILTERS, _SEEN, _EXC_CLS
if regex or search_msg:
real_warn = warnings.warn
def our_warn(msg, *arg, **kw):
-
if isinstance(msg, _EXC_CLS):
exception = type(msg)
msg = str(msg)
"""
- with _expect_warnings(except_cls, [".*"], squelch_other_warnings=True):
+ with _expect_warnings_sqla_only(except_cls, [".*"]):
return callable_(*args, **kwargs)
rather than regex.match().
"""
- with _expect_warnings(
+ with _expect_warnings_sqla_only(
except_cls,
[msg],
search_msg=True,
regex=False,
- squelch_other_warnings=True,
):
return callable_(*args, **kwargs)
def _assert_raises(
except_cls, callable_, args, kwargs, msg=None, check_context=False
):
-
with _expect_raises(except_cls, msg, check_context) as ec:
callable_(*args, **kwargs)
return ec.error
return result
def assert_sql(self, db, callable_, rules):
-
newrules = []
for rule in rules:
if isinstance(rule, dict):
from typing import Callable
from typing import Coroutine
from typing import Optional
+from typing import TYPE_CHECKING
from typing import TypeVar
from .langhelpers import memoized_property
from .. import exc
from ..util.typing import Protocol
+from ..util.typing import TypeGuard
_T = TypeVar("_T")
self.gr_context = driver.gr_context
+_T_co = TypeVar("_T_co", covariant=True)
+
+if TYPE_CHECKING:
+
+ def iscoroutine(
+ awaitable: Awaitable[_T_co],
+ ) -> TypeGuard[Coroutine[Any, Any, _T_co]]:
+ ...
+
+else:
+ iscoroutine = asyncio.iscoroutine
+
+
+def _safe_cancel_awaitable(awaitable: Awaitable[Any]) -> None:
+ # https://docs.python.org/3/reference/datamodel.html#coroutine.close
+
+ if iscoroutine(awaitable):
+ awaitable.close()
+
+
def await_only(awaitable: Awaitable[_T]) -> _T:
"""Awaits an async function in a sync method.
# this is called in the context greenlet while running fn
current = getcurrent()
if not isinstance(current, _AsyncIoGreenlet):
+ _safe_cancel_awaitable(awaitable)
+
raise exc.MissingGreenlet(
"greenlet_spawn has not been called; can't call await_only() "
"here. Was IO attempted in an unexpected place?"
if not isinstance(current, _AsyncIoGreenlet):
loop = get_event_loop()
if loop.is_running():
+
+ _safe_cancel_awaitable(awaitable)
+
raise exc.MissingGreenlet(
"greenlet_spawn has not been called and asyncio event "
"loop is already running; can't call await_fallback() here. "
@testing.requires.cextensions
def test_cycles_in_row(self):
-
tup = result.result_tuple(["a", "b", "c"])
@profile_memory()
@testing.emits_warning()
@profile_memory()
def go():
-
# execute with a non-unicode object. a warning is emitted,
# this warning shouldn't clog up memory.
t1_mapper = self.mapper_registry.map_imperatively(T1, t1)
- @testing.emits_warning()
+ @testing.emits_warning(
+ r"This declarative base", r"Property .* being replaced"
+ )
@profile_memory()
def go():
class T2:
s = table2.select()
sess = session()
with testing.expect_deprecated(
- "Implicit coercion of SELECT and " "textual SELECT constructs"
+ "Implicit coercion of SELECT and textual SELECT constructs",
+ "An alias is being generated automatically",
+ assert_=False,
):
sess.query(Foo).join(s, Foo.bars).all()
sess.rollback()
@testing.provide_metadata
def test_optimized_get(self):
-
Base = declarative_base(metadata=self.metadata)
class Employee(Base):
):
await_only(to_await)
- # ensure no warning
+ # existing awaitable is done
+ with expect_raises(RuntimeError):
+ await greenlet_spawn(await_fallback, to_await)
+
+ # no warning for a new one...
+ to_await = run1()
await greenlet_spawn(await_fallback, to_await)
@async_test
):
await greenlet_spawn(go)
- await to_await
+ with expect_raises(RuntimeError):
+ await to_await
@async_test
async def test_await_only_error(self):
):
await greenlet_spawn(go)
- await to_await
+ with expect_raises(RuntimeError):
+ await to_await
@async_test
async def test_contextvars(self):
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import Table
-from sqlalchemy import testing
from sqlalchemy import true
from sqlalchemy.testing import eq_
+from sqlalchemy.testing import expect_warnings
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
__only_on__ = "mysql", "mariadb"
__backend__ = True
- @testing.emits_warning()
def test_is_boolean_symbols_despite_no_native(self, connection):
+ with expect_warnings("Datatype BOOL does not support CAST"):
+ is_(
+ connection.scalar(select(cast(true().is_(true()), Boolean))),
+ True,
+ )
- is_(
- connection.scalar(select(cast(true().is_(true()), Boolean))),
- True,
- )
-
- is_(
- connection.scalar(select(cast(true().is_not(true()), Boolean))),
- False,
- )
+ with expect_warnings("Datatype BOOL does not support CAST"):
+ is_(
+ connection.scalar(
+ select(cast(true().is_not(true()), Boolean))
+ ),
+ False,
+ )
- is_(
- connection.scalar(select(cast(false().is_(false()), Boolean))),
- True,
- )
+ with expect_warnings("Datatype BOOL does not support CAST"):
+ is_(
+ connection.scalar(select(cast(false().is_(false()), Boolean))),
+ True,
+ )
class MatchTest(fixtures.TablesTest):
select(1).compile(dialect=e1.dialect), (), {}
)
+ @testing.emits_warning("The garbage collector is trying to clean up")
def test_execute_events(self):
stmts = []
eng_copy = copy.copy(eng)
eng_copy.dispose(close=close)
+
copy_conn = eng_copy.connect()
dbapi_conn_two = copy_conn.connection.dbapi_connection
else:
is_(dbapi_conn_one, conn.connection.dbapi_connection)
+ conn.close()
+ copy_conn.close()
+
def test_retval_flag(self):
canary = []
conn.connection.invalidate(soft=True)
conn.close()
- conn = e.connect()
- eq_(conn.info["boom"], "one")
+ with e.connect() as conn:
+ eq_(conn.info["boom"], "one")
def test_connect_do_connect_info_there_after_invalidate(self):
# test that info is maintained after the do_connect()
eq_(conn.info["boom"], "one")
conn.connection.invalidate()
- conn = e.connect()
- eq_(conn.info["boom"], "one")
+
+ with e.connect() as conn:
+ eq_(conn.info["boom"], "one")
class SetInputSizesTest(fixtures.TablesTest):
)
@testing.combinations((True,), (False,))
+ @testing.emits_warning("The garbage collector")
def test_userspace_disconnectionerror_weakref_finalizer(self, detach_gced):
dbapi, pool = self._queuepool_dbapi_fixture(
pool_size=1, max_overflow=2, _is_asyncio=detach_gced
not_closed_dbapi_conn = conn.dbapi_connection
del conn
+
gc_collect()
if detach_gced:
eq_(not_closed_dbapi_conn.mock_calls, [])
else:
# new connection reset and returned to pool
+ # this creates a gc-level warning that is not easy to pin down,
+ # hence we use the testing.emits_warning() decorator just to squash
+ # it
eq_(not_closed_dbapi_conn.mock_calls, [call.rollback()])
@testing.requires.timing_intensive
(exc.StatementError, exc.MissingGreenlet)
):
a1.b = b2
+
else:
a1.b = b2
m1 = mock.Mock()
with testing.expect_deprecated(
- "The ``query_chooser`` parameter is deprecated; please use"
+ "The ``query_chooser`` parameter is deprecated; please use",
):
s = ShardedSession(
shard_chooser=m1.shard_chooser,
- id_chooser=m1.id_chooser,
+ identity_chooser=m1.identity_chooser,
query_chooser=m1.query_chooser,
)
r"non-schema SQLAlchemy expression object; ",
r"Attribute 'y' on class <class .*Foo5.* appears to be a "
r"non-schema SQLAlchemy expression object; ",
- raise_on_any_unexpected=True,
):
class Foo5(Base):
if not combine_on_b and not omit_from_statements:
ctx = expect_warnings(
"Implicitly combining column a.extra with column b.extra",
- raise_on_any_unexpected=True,
)
else:
ctx = contextlib.nullcontext()
return testing.expect_warnings(
r"An alias is being generated automatically against joined entity "
r"Mapper\[%s\] due to overlapping tables" % (arg,),
- raise_on_any_unexpected=True,
)
"'bs' was None;",
"Attribute keyed dictionary value for attribute "
"'B.a' was None;",
- raise_on_any_unexpected=True,
):
a1 = A(bs={"k1": b1, "k2": b2})
sess.add(a1)
with expect_warnings(
"Attribute keyed dictionary value for attribute "
"'unknown relationship' was None;",
- raise_on_any_unexpected=True,
):
eq_(a1.bs, {None: b2})
with expect_warnings(
"Attribute keyed dictionary value for attribute "
"'B.a' was None;",
- raise_on_any_unexpected=True,
):
b1.a = a1
else:
with expect_deprecated(
rf"The column_property.{paramname} parameter is deprecated "
r"for column_property\(\)",
- raise_on_any_unexpected=True,
):
column_property(column("q"), **{paramname: value})
r"for column_property\(\)",
r"The column_property.kw_only parameter is deprecated "
r"for column_property\(\)",
- raise_on_any_unexpected=True,
):
class MyClass(dc_decl_base):
s = addresses.select()
sess = fixture_session()
with testing.expect_deprecated(
- "Implicit coercion of SELECT and " "textual SELECT constructs"
+ "Implicit coercion of SELECT and textual SELECT constructs",
+ "An alias is being generated automatically against joined entity",
):
self.assert_compile(
sess.query(User).join(s, User.addresses),
def go():
with testing.expect_deprecated(
"The AliasOption object is not necessary for entities to be "
- "matched up to a query"
+ "matched up to a query",
):
result = (
q.options(
def go():
with testing.expect_deprecated(
- r"Using the Query.instances\(\) method without a context"
+ r"The Query.instances\(\) method is deprecated",
+ r"Using the Query.instances\(\) method without a context",
):
result = list(
q.options(contains_eager(User.addresses)).instances(
def go():
with testing.expect_deprecated(
- r"Using the Query.instances\(\) method without a context"
+ r"The Query.instances\(\) method is deprecated",
+ r"Using the Query.instances\(\) method without a context",
):
result = list(
q.options(contains_eager(User.addresses)).instances(
r"Using the Query.instances\(\) method without a context",
r"The Query.instances\(\) method is deprecated and will be "
r"removed in a future release.",
- raise_on_any_unexpected=True,
):
result = list(
q.options(
r"Using the Query.instances\(\) method without a context",
r"The Query.instances\(\) method is deprecated and will be "
r"removed in a future release.",
- raise_on_any_unexpected=True,
):
result = list(
q.options(
def go():
eq_(u1.addresses[0].user, u1)
- with testing.expect_warnings(raise_on_any_unexpected=True):
+ with testing.expect_warnings():
self.assert_sql_execution(
testing.db,
go,
def go():
eq_(u1.addresses[0].user, u1)
- with testing.expect_warnings(raise_on_any_unexpected=True):
+ with testing.expect_warnings():
self.assert_sql_execution(
testing.db,
go,
r"Mapper\[Manager\(managers\)\] due to overlapping",
"An alias is being generated automatically against joined entity "
r"Mapper\[Boss\(boss\)\] due to overlapping",
- raise_on_any_unexpected=True,
):
self.assert_compile(
q,
from sqlalchemy.orm.interfaces import ONETOMANY
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
-from sqlalchemy.testing import assert_warns_message
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
from sqlalchemy.testing import expect_raises_message
add_bsub2_a_viewonly=False,
add_b_a_overlaps=None,
):
-
Base = self.mapper_registry.generate_base()
class A(Base):
-
__tablename__ = "a"
id = Column(Integer, primary_key=True)
a_members = relationship("AMember", backref="a")
class AMember(Base):
-
__tablename__ = "a_member"
a_id = Column(Integer, ForeignKey("a.id"), primary_key=True)
# however, *no* warning should be emitted otherwise.
class BSub1(B):
-
if add_bsub1_a:
a = relationship("A")
__mapper_args__ = {"polymorphic_identity": "bsub1"}
class BSub2(B):
-
if add_bsub2_a_viewonly:
a = relationship("A", viewonly=True)
return A, AMember, B, BSub1, BSub2
def _fixture_two(self, setup_backrefs=False, setup_overlaps=False):
-
Base = self.mapper_registry.generate_base()
# purposely using the comma to make sure parsing the comma works
@testing.provide_metadata
def test_simple_warn(self):
- assert_warns_message(
- exc.SAWarning,
+ with expect_warnings(
r"relationship '(?:Child.parent|Parent.children)' will copy "
r"column parent.id to column child.parent_id, which conflicts "
r"with relationship\(s\): '(?:Parent.children|Child.parent)' "
- r"\(copies parent.id to child.parent_id\).",
- self._fixture_two,
- setup_backrefs=False,
- )
+ r"\(copies parent.id to child.parent_id\)."
+ ):
+ self._fixture_two(setup_backrefs=False)
@testing.combinations((True,), (False,), argnames="set_overlaps")
def test_fixture_five(self, metadata, set_overlaps):
@testing.provide_metadata
def test_double_rel_same_mapper_warns(self):
- assert_warns_message(
- exc.SAWarning,
+ with expect_warnings(
r"relationship 'Parent.child[12]' will copy column parent.id to "
r"column child.parent_id, which conflicts with relationship\(s\): "
- r"'Parent.child[12]' \(copies parent.id to child.parent_id\)",
- self._fixture_three,
- use_same_mappers=True,
- setup_overlaps=False,
- )
+ r"'Parent.child[12]' \(copies parent.id to child.parent_id\)"
+ ):
+ self._fixture_three(use_same_mappers=True, setup_overlaps=False)
@testing.provide_metadata
def test_double_rel_same_mapper_overlaps_works(self):
@testing.provide_metadata
def test_warn_one(self):
- assert_warns_message(
- exc.SAWarning,
+ with expect_warnings(
r"relationship '(?:BSub1.a|BSub2.a_member|B.a)' will copy column "
- r"(?:a.id|a_member.a_id) to column b.a_id",
- self._fixture_one,
- add_b_a=True,
- add_bsub1_a=True,
- )
+ r"(?:a.id|a_member.a_id) to column b.a_id"
+ ):
+ self._fixture_one(add_b_a=True, add_bsub1_a=True)
@testing.provide_metadata
def test_warn_two(self):
- assert_warns_message(
- exc.SAWarning,
+ with expect_warnings(
r"relationship '(?:BSub1.a|B.a_member)' will copy column "
- r"(?:a.id|a_member.a_id) to column b.a_id",
- self._fixture_one,
- add_b_amember=True,
- add_bsub1_a=True,
- )
+ r"(?:a.id|a_member.a_id) to column b.a_id"
+ ):
+ self._fixture_one(add_b_amember=True, add_bsub1_a=True)
@testing.provide_metadata
def test_warn_three(self):
- assert_warns_message(
- exc.SAWarning,
- r"relationship '(?:BSub1.a|B.a_member|B.a)' will copy column "
- r"(?:a.id|a_member.a_id) to column b.a_id",
- self._fixture_one,
- add_b_amember=True,
- add_bsub1_a=True,
- add_b_a=True,
- )
+ with expect_warnings(
+ r"relationship '(?:BSub1.a|B.a_member|BSub2.a_member|B.a)' "
+ r"will copy column (?:a.id|a_member.a_id) to column b.a_id",
+ ):
+ self._fixture_one(
+ add_b_amember=True, add_bsub1_a=True, add_b_a=True
+ )
@testing.provide_metadata
def test_warn_four(self):
- assert_warns_message(
- exc.SAWarning,
+ with expect_warnings(
r"relationship '(?:B.a|BSub2.a_member|B.a)' will copy column "
- r"(?:a.id|a_member.a_id) to column b.a_id",
- self._fixture_one,
- add_bsub2_a_viewonly=True,
- add_b_a=True,
- )
+ r"(?:a.id|a_member.a_id) to column b.a_id"
+ ):
+ self._fixture_one(add_bsub2_a_viewonly=True, add_b_a=True)
@testing.provide_metadata
def test_works_one(self):
},
)
- assert_warns_message(
- exc.SAWarning,
+ with expect_warnings(
r"relationship .* will copy column .* to column "
- r"employee_t.company_id, which conflicts with relationship\(s\)",
- configure_mappers,
- )
+ r"employee_t.company_id, which conflicts with relationship\(s\)"
+ ):
+ configure_mappers()
def test_annotated_no_overwriting(self):
Employee, Company, employee_t, company_t = (
argnames="argname, arg",
)
def test_invalid_string_args(self, registry, argname, arg):
-
kw = {argname: arg}
Base = registry.generate_base()
)
def test_render_lazyload(self):
-
A = self.classes.A
sess = fixture_session()
a1 = sess.query(A).filter(A.name == "a1").first()
delete,
legacy_inactive_history_style,
):
-
if delete:
assert not backref, "delete and backref are mutually exclusive"
from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
from sqlalchemy.testing import expect_raises_message
+from sqlalchemy.testing import expect_warnings
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
from sqlalchemy.testing import is_false
)
def test_no_string_execute(self, connection):
-
with Session(bind=connection) as sess:
with expect_raises_message(
sa.exc.ArgumentError,
self.mapper_registry.map_imperatively(Address, addresses)
with Session(testing.db) as sess:
-
sess.add(User(name="u1"))
sess.commit()
def evt(mapper, conn, instance):
instance.addresses.append(Address(email="x1"))
- self._test(evt, "collection append")
+ self._test(evt, "collection append", "related attribute set")
def test_o2m_cascade_remove(self):
def evt(mapper, conn, instance):
del instance.addresses[0]
- self._test(evt, "collection remove")
+ self._test(evt, "collection remove", "related attribute set")
def test_m2o_cascade_add(self):
User = self.classes.User
instance.addresses[0].user = User(name="u2")
with expect_raises_message(orm_exc.FlushError, ".*Over 100"):
- self._test(evt, "related attribute set")
+ self._test(
+ evt,
+ "related attribute set",
+ "collection remove",
+ "collection append",
+ )
def test_m2o_cascade_remove(self):
def evt(mapper, conn, instance):
a1 = instance.addresses[0]
del a1.user
- self._test(evt, "related attribute delete")
+ self._test(evt, "related attribute delete", "collection remove")
def test_plain_add(self):
Address = self.classes.Address
):
self._test(evt, r"Session.delete\(\)")
- def _test(self, fn, method):
+ def _test(self, fn, *methods):
User = self.classes.User
Address = self.classes.Address
u1 = User(name="u1", addresses=[Address(name="a1")])
s.add(u1)
- assert_warns_message(
- sa.exc.SAWarning, "Usage of the '%s'" % method, s.commit
- )
+
+ with expect_warnings(
+ *[f"Usage of the '{method}'" for method in methods]
+ ):
+ s.commit()
"bar.foo_id, which conflicts with relationship(s): 'Foo.bars' "
"(copies foo.id to bar.foo_id). "
),
- raise_on_any_unexpected=True,
):
decl_base.registry.configure()
"invoked automatically in response to a user-initiated "
"operation.)"
),
- raise_on_any_unexpected=True,
):
FooAlias = aliased(Foo)
assert hasattr(FooAlias, "bars")
"invoked automatically in response to a user-initiated "
"operation.)"
),
- raise_on_any_unexpected=True,
):
foo = Foo()
assert hasattr(foo, "bars")
"process, which was invoked automatically in response to a "
"user-initiated operation.)"
),
- raise_on_any_unexpected=True,
):
sess.execute(select(Foo))
def test_where_empty(self):
table1 = self.tables.mytable
- with expect_deprecated():
+ with expect_deprecated(
+ r"Invoking and_\(\) without arguments is deprecated"
+ ):
self.assert_compile(
table1.delete().where(and_()), "DELETE FROM mytable"
)
- with expect_deprecated():
+ with expect_deprecated(
+ r"Invoking or_\(\) without arguments is deprecated"
+ ):
self.assert_compile(
table1.delete().where(or_()), "DELETE FROM mytable"
)
):
return expect_warnings(
"Batches were downgraded",
- raise_on_any_unexpected=True,
)
return contextlib.nullcontext()
):
with expect_warnings(
"Batches were downgraded for sorted INSERT",
- raise_on_any_unexpected=True,
):
result = connection.execute(stmt, data)
else:
with expect_warnings(
r".*but has no Python-side or server-side default ",
- raise_on_any_unexpected=True,
):
with expect_raises(exc.IntegrityError):
connection.execute(
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
