From 2665a0c4cb3e94e6545d0b9bbcbcc39ccffebaba Mon Sep 17 00:00:00 2001 From: Mike Bayer Date: Thu, 8 Oct 2020 15:20:48 -0400 Subject: generalize scoped_session proxying and apply to asyncio elements Reworked the proxy creation used by scoped_session() to be based on fully copied code with augmented docstrings and moved it into langhelpers. asyncio session, engine, connection can now take advantage of it so that all non-async methods are availble. Overall implementation of most important accessors / methods on AsyncConnection, etc. , including awaitable versions of invalidate, execution_options, etc. In order to support an event dispatcher on the async classes while still allowing them to hold __slots__, make some adjustments to the event system to allow that to be present, at least rudimentally. Fixes: #5628 Change-Id: I5eb6929fc1e4fdac99e4b767dcfd49672d56e2b2 --- test/ext/asyncio/test_engine_py3k.py | 182 ++++++++++++++++++++++++++++++++++ test/ext/asyncio/test_session_py3k.py | 58 +++++++++++ 2 files changed, 240 insertions(+) (limited to 'test/ext') diff --git a/test/ext/asyncio/test_engine_py3k.py b/test/ext/asyncio/test_engine_py3k.py index 7c7d90e21..83987b06f 100644 --- a/test/ext/asyncio/test_engine_py3k.py +++ b/test/ext/asyncio/test_engine_py3k.py @@ -2,6 +2,7 @@ import asyncio from sqlalchemy import Column from sqlalchemy import delete +from sqlalchemy import event from sqlalchemy import exc from sqlalchemy import func from sqlalchemy import Integer @@ -9,13 +10,19 @@ from sqlalchemy import select from sqlalchemy import String from sqlalchemy import Table from sqlalchemy import testing +from sqlalchemy import text from sqlalchemy import union_all from sqlalchemy.ext.asyncio import create_async_engine +from sqlalchemy.ext.asyncio import engine as _async_engine from sqlalchemy.ext.asyncio import exc as asyncio_exc from sqlalchemy.testing import async_test from sqlalchemy.testing import eq_ from sqlalchemy.testing import fixtures +from sqlalchemy.testing import is_ +from sqlalchemy.testing import is_not +from sqlalchemy.testing import mock from sqlalchemy.testing.asyncio import assert_raises_message_async +from sqlalchemy.util.concurrency import greenlet_spawn class EngineFixture(fixtures.TablesTest): @@ -50,6 +57,117 @@ class EngineFixture(fixtures.TablesTest): class AsyncEngineTest(EngineFixture): __backend__ = True + def test_proxied_attrs_engine(self, async_engine): + sync_engine = async_engine.sync_engine + + is_(async_engine.url, sync_engine.url) + is_(async_engine.pool, sync_engine.pool) + is_(async_engine.dialect, sync_engine.dialect) + eq_(async_engine.name, sync_engine.name) + eq_(async_engine.driver, sync_engine.driver) + eq_(async_engine.echo, sync_engine.echo) + + def test_clear_compiled_cache(self, async_engine): + async_engine.sync_engine._compiled_cache["foo"] = "bar" + eq_(async_engine.sync_engine._compiled_cache["foo"], "bar") + async_engine.clear_compiled_cache() + assert "foo" not in async_engine.sync_engine._compiled_cache + + def test_execution_options(self, async_engine): + a2 = async_engine.execution_options(foo="bar") + assert isinstance(a2, _async_engine.AsyncEngine) + eq_(a2.sync_engine._execution_options, {"foo": "bar"}) + eq_(async_engine.sync_engine._execution_options, {}) + + """ + + attr uri, pool, dialect, engine, name, driver, echo + methods clear_compiled_cache, update_execution_options, + execution_options, get_execution_options, dispose + + """ + + @async_test + async def test_proxied_attrs_connection(self, async_engine): + conn = await async_engine.connect() + + sync_conn = conn.sync_connection + + is_(conn.engine, async_engine) + is_(conn.closed, sync_conn.closed) + is_(conn.dialect, async_engine.sync_engine.dialect) + eq_(conn.default_isolation_level, sync_conn.default_isolation_level) + + @async_test + async def test_invalidate(self, async_engine): + conn = await async_engine.connect() + + is_(conn.invalidated, False) + + connection_fairy = await conn.get_raw_connection() + is_(connection_fairy.is_valid, True) + dbapi_connection = connection_fairy.connection + + await conn.invalidate() + assert dbapi_connection._connection.is_closed() + + new_fairy = await conn.get_raw_connection() + is_not(new_fairy.connection, dbapi_connection) + is_not(new_fairy, connection_fairy) + is_(new_fairy.is_valid, True) + is_(connection_fairy.is_valid, False) + + @async_test + async def test_get_dbapi_connection_raise(self, async_engine): + + conn = await async_engine.connect() + + with testing.expect_raises_message( + exc.InvalidRequestError, + "AsyncConnection.connection accessor is not " + "implemented as the attribute", + ): + conn.connection + + @async_test + async def test_get_raw_connection(self, async_engine): + + conn = await async_engine.connect() + + pooled = await conn.get_raw_connection() + is_(pooled, conn.sync_connection.connection) + + @async_test + async def test_isolation_level(self, async_engine): + conn = await async_engine.connect() + + sync_isolation_level = await greenlet_spawn( + conn.sync_connection.get_isolation_level + ) + isolation_level = await conn.get_isolation_level() + + eq_(isolation_level, sync_isolation_level) + + await conn.execution_options(isolation_level="SERIALIZABLE") + isolation_level = await conn.get_isolation_level() + + eq_(isolation_level, "SERIALIZABLE") + + @async_test + async def test_dispose(self, async_engine): + c1 = await async_engine.connect() + c2 = await async_engine.connect() + + await c1.close() + await c2.close() + + p1 = async_engine.pool + eq_(async_engine.pool.checkedin(), 2) + + await async_engine.dispose() + eq_(async_engine.pool.checkedin(), 0) + is_not(p1, async_engine.pool) + @async_test async def test_init_once_concurrency(self, async_engine): c1 = async_engine.connect() @@ -169,6 +287,70 @@ class AsyncEngineTest(EngineFixture): ) +class AsyncEventTest(EngineFixture): + """The engine events all run in their normal synchronous context. + + we do not provide an asyncio event interface at this time. + + """ + + __backend__ = True + + @async_test + async def test_no_async_listeners(self, async_engine): + with testing.expect_raises_message( + NotImplementedError, + "asynchronous events are not implemented " + "at this time. Apply synchronous listeners to the " + "AsyncEngine.sync_engine or " + "AsyncConnection.sync_connection attributes.", + ): + event.listen(async_engine, "before_cursor_execute", mock.Mock()) + + conn = await async_engine.connect() + + with testing.expect_raises_message( + NotImplementedError, + "asynchronous events are not implemented " + "at this time. Apply synchronous listeners to the " + "AsyncEngine.sync_engine or " + "AsyncConnection.sync_connection attributes.", + ): + event.listen(conn, "before_cursor_execute", mock.Mock()) + + @async_test + async def test_sync_before_cursor_execute_engine(self, async_engine): + canary = mock.Mock() + + event.listen(async_engine.sync_engine, "before_cursor_execute", canary) + + async with async_engine.connect() as conn: + sync_conn = conn.sync_connection + await conn.execute(text("select 1")) + + eq_( + canary.mock_calls, + [mock.call(sync_conn, mock.ANY, "select 1", (), mock.ANY, False)], + ) + + @async_test + async def test_sync_before_cursor_execute_connection(self, async_engine): + canary = mock.Mock() + + async with async_engine.connect() as conn: + sync_conn = conn.sync_connection + + event.listen( + async_engine.sync_engine, "before_cursor_execute", canary + ) + await conn.execute(text("select 1")) + + eq_( + canary.mock_calls, + [mock.call(sync_conn, mock.ANY, "select 1", (), mock.ANY, False)], + ) + + class AsyncResultTest(EngineFixture): @testing.combinations( (None,), ("scalars",), ("mappings",), argnames="filter_" diff --git a/test/ext/asyncio/test_session_py3k.py b/test/ext/asyncio/test_session_py3k.py index e8caaca3e..a3b8add67 100644 --- a/test/ext/asyncio/test_session_py3k.py +++ b/test/ext/asyncio/test_session_py3k.py @@ -1,3 +1,4 @@ +from sqlalchemy import event from sqlalchemy import exc from sqlalchemy import func from sqlalchemy import select @@ -9,6 +10,7 @@ from sqlalchemy.orm import selectinload from sqlalchemy.testing import async_test from sqlalchemy.testing import eq_ from sqlalchemy.testing import is_ +from sqlalchemy.testing import mock from ...orm import _fixtures @@ -139,6 +141,27 @@ class AsyncSessionTransactionTest(AsyncFixture): eq_(await outer_conn.scalar(select(func.count(User.id))), 1) + @async_test + async def test_delete(self, async_session): + User = self.classes.User + + async with async_session.begin(): + u1 = User(name="u1") + + async_session.add(u1) + + await async_session.flush() + + conn = await async_session.connection() + + eq_(await conn.scalar(select(func.count(User.id))), 1) + + async_session.delete(u1) + + await async_session.flush() + + eq_(await conn.scalar(select(func.count(User.id))), 0) + @async_test async def test_flush(self, async_session): User = self.classes.User @@ -198,3 +221,38 @@ class AsyncSessionTransactionTest(AsyncFixture): is_(new_u_merged, u1) eq_(u1.name, "new u1") + + +class AsyncEventTest(AsyncFixture): + """The engine events all run in their normal synchronous context. + + we do not provide an asyncio event interface at this time. + + """ + + __backend__ = True + + @async_test + async def test_no_async_listeners(self, async_session): + with testing.expect_raises( + NotImplementedError, + "NotImplementedError: asynchronous events are not implemented " + "at this time. Apply synchronous listeners to the " + "AsyncEngine.sync_engine or " + "AsyncConnection.sync_connection attributes.", + ): + event.listen(async_session, "before_flush", mock.Mock()) + + @async_test + async def test_sync_before_commit(self, async_session): + canary = mock.Mock() + + event.listen(async_session.sync_session, "before_commit", canary) + + async with async_session.begin(): + pass + + eq_( + canary.mock_calls, + [mock.call(async_session.sync_session)], + ) -- cgit v1.2.1