:ticket:`1390`
+.. _deprecation_20_mode:
+
+SQLAlchemy 2.0 Deprecations Mode
+---------------------------------
+
+One of the primary goals of the 1.4 release is to provide a "transitional"
+release so that applications may migrate to SQLAlchemy 2.0 gradually. Towards
+this end, a primary feature in release 1.4 is "2.0 deprecations mode", which is
+a series of deprecation warnings that emit against every detectable API pattern
+which will work differently in version 2.0. The warnings all make use of the
+:class:`_exc.RemovedIn20Warning` class. As these warnings affect foundational
+patterns including the :func:`_sql.select` and :class:`_engine.Engine` constructs, even
+simple applications can generate a lot of warnings until appropriate API
+changes are made. The warning mode is therefore turned off by default until
+the developer enables the environment variable ``SQLALCHEMY_WARN_20=1``.
+
+Given the example program below::
+
+ from sqlalchemy import column
+ from sqlalchemy import create_engine
+ from sqlalchemy import select
+ from sqlalchemy import table
+
+
+ engine = create_engine("sqlite://")
+
+ engine.execute("CREATE TABLE foo (id integer)")
+ engine.execute("INSERT INTO foo (id) VALUES (1)")
+
+
+ foo = table("foo", column("id"))
+ result = engine.execute(select([foo.c.id]))
+
+ print(result.fetchall())
+
+The above program uses several patterns that many users will already identify
+as "legacy", namely the use of the :meth:`_engine.Engine.execute` method
+that's part of the :ref:`connectionlesss execution <dbengine_implicit>`
+system. When we run the above program against 1.4, it returns a single line::
+
+ $ python test3.py
+ [(1,)]
+
+To enable "2.0 deprecations mode", we enable the ``SQLALCHEMY_WARN_20=1``
+variable::
+
+ SQLALCHEMY_WARN_20=1 python test3.py
+
+**IMPORTANT** - older versions of Python may not emit deprecation warnings
+by default. To guarantee deprecation warnings, use a `warnings filter`_
+that ensures warnings are printed::
+
+ SQLALCHEMY_WARN_20=1 python -W always::DeprecationWarning test3.py
+
+.. _warnings filter: https://docs.python.org/3/library/warnings.html#the-warnings-filter
+
+With warnings turned on, our program now has a lot to say::
+
+ $ SQLALCHEMY_WARN_20=1 python2 -W always::DeprecationWarning test3.py
+ test3.py:9: RemovedIn20Warning: The Engine.execute() function/method is considered legacy as of the 1.x series of SQLAlchemy and will be removed in 2.0. All statement execution in SQLAlchemy 2.0 is performed by the Connection.execute() method of Connection, or in the ORM by the Session.execute() method of Session. (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9) (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9)
+ engine.execute("CREATE TABLE foo (id integer)")
+ /home/classic/dev/sqlalchemy/lib/sqlalchemy/engine/base.py:2856: RemovedIn20Warning: Passing a string to Connection.execute() is deprecated and will be removed in version 2.0. Use the text() construct, or the Connection.exec_driver_sql() method to invoke a driver-level SQL string. (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9)
+ return connection.execute(statement, *multiparams, **params)
+ /home/classic/dev/sqlalchemy/lib/sqlalchemy/engine/base.py:1639: RemovedIn20Warning: The current statement is being autocommitted using implicit autocommit.Implicit autocommit will be removed in SQLAlchemy 2.0. Use the .begin() method of Engine or Connection in order to use an explicit transaction for DML and DDL statements. (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9)
+ self._commit_impl(autocommit=True)
+ test3.py:10: RemovedIn20Warning: The Engine.execute() function/method is considered legacy as of the 1.x series of SQLAlchemy and will be removed in 2.0. All statement execution in SQLAlchemy 2.0 is performed by the Connection.execute() method of Connection, or in the ORM by the Session.execute() method of Session. (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9) (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9)
+ engine.execute("INSERT INTO foo (id) VALUES (1)")
+ /home/classic/dev/sqlalchemy/lib/sqlalchemy/engine/base.py:2856: RemovedIn20Warning: Passing a string to Connection.execute() is deprecated and will be removed in version 2.0. Use the text() construct, or the Connection.exec_driver_sql() method to invoke a driver-level SQL string. (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9)
+ return connection.execute(statement, *multiparams, **params)
+ /home/classic/dev/sqlalchemy/lib/sqlalchemy/engine/base.py:1639: RemovedIn20Warning: The current statement is being autocommitted using implicit autocommit.Implicit autocommit will be removed in SQLAlchemy 2.0. Use the .begin() method of Engine or Connection in order to use an explicit transaction for DML and DDL statements. (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9)
+ self._commit_impl(autocommit=True)
+ /home/classic/dev/sqlalchemy/lib/sqlalchemy/sql/selectable.py:4271: RemovedIn20Warning: The legacy calling style of select() is deprecated and will be removed in SQLAlchemy 2.0. Please use the new calling style described at select(). (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9) (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9)
+ return cls.create_legacy_select(*args, **kw)
+ test3.py:14: RemovedIn20Warning: The Engine.execute() function/method is considered legacy as of the 1.x series of SQLAlchemy and will be removed in 2.0. All statement execution in SQLAlchemy 2.0 is performed by the Connection.execute() method of Connection, or in the ORM by the Session.execute() method of Session. (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9) (Background on SQLAlchemy 2.0 at: http://sqlalche.me/e/b8d9)
+ result = engine.execute(select([foo.c.id]))
+ [(1,)]
+
+With the above guidance, we can migrate our program to use 2.0 styles, and
+as a bonus our program is much clearer::
+
+ from sqlalchemy import column
+ from sqlalchemy import create_engine
+ from sqlalchemy import select
+ from sqlalchemy import table
+ from sqlalchemy import text
+
+
+ engine = create_engine("sqlite://")
+
+ # don't rely on autocommit for DML and DDL
+ with engine.begin() as connection:
+ # use connection.execute(), not engine.execute()
+ # use the text() construct to execute textual SQL
+ connection.execute(text("CREATE TABLE foo (id integer)"))
+ connection.execute(text("INSERT INTO foo (id) VALUES (1)"))
+
+
+ foo = table("foo", column("id"))
+
+ with engine.connect() as connection:
+ # use connection.execute(), not engine.execute()
+ # select() now accepts column / table expressions positionally
+ result = connection.execute(select(foo.c.id))
+
+ print(result.fetchall())
+
+
+The goal of "2.0 deprecations mode" is that a program which runs with no
+:class:`_exc.RemovedIn20Warning` warnings with "2.0 deprecations mode" turned
+on is then ready to run in SQLAlchemy 2.0.
+
+
+.. seealso::
+
+ :ref:`migration_20_toplevel`
+
+
+
API and Behavioral Changes - Core
==================================
well as providing for the initial real-world adoption of the new
architectures.
-* A new deprecation class :class:`.exc.RemovedIn20Warning` is added, which
- subclasses :class:`.exc.SADeprecationWarning`. Applications and their test
+* A new deprecation class :class:`_exc.RemovedIn20Warning` is added, which
+ subclasses :class:`_exc.SADeprecationWarning`. Applications and their test
suites can opt to enable or disable reporting of the
- :class:`.exc.RemovedIn20Warning` warning as needed. To some extent, the
- :class:`.exc.RemovedIn20Warning` deprecation class is analogous to the ``-3``
+ :class:`_exc.RemovedIn20Warning` warning as needed, by setting the
+ environment variable ``SQLALCHEMY_WARN_20=1`` **before** the program
+ runs. To some extent, the
+ :class:`_exc.RemovedIn20Warning` deprecation class is analogous to the ``-3``
flag available on Python 2 which reports on future Python 3
- incompatibilities.
+ incompatibilities. See :ref:`deprecation_20_mode` for background
+ on turning this on.
-* APIs which emit :class:`.exc.RemovedIn20Warning` should always feature a new
+* APIs which emit :class:`_exc.RemovedIn20Warning` should always feature a new
1.4-compatible usage pattern that applications can migrate towards. This
pattern will then be fully compatible with SQLAlchemy 2.0. In this way,
an application can gradually adjust all of its 1.4-style code to work fully
--- /dev/null
+.. change::
+ :tags: engine
+ :tickets: 4846
+
+ "Implicit autocommit", which is the COMMIT that occurs when a DML or DDL
+ statement is emitted on a connection, is deprecated and won't be part of
+ SQLAlchemy 2.0. A 2.0-style warning is emitted when autocommit takes
+ effect, so that the calling code may be adjusted to use an explicit
+ transaction.
+
+ As part of this change, DDL methods such as
+ :meth:`_schema.MetaData.create_all` when used against an
+ :class:`_engine.Engine` will run the operation in a BEGIN block if one is
+ not started already.
+
+ .. seealso::
+
+ :ref:`deprecation_20_mode`
+
--- /dev/null
+.. change::
+ :tags: bug, mysql
+
+ The MySQL and MariaDB dialects now query from the information_schema.tables
+ system view in order to determine if a particular table exists or not.
+ Previously, the "DESCRIBE" command was used with an exception catch to
+ detect non-existent, which would have the undesirable effect of emitting a
+ ROLLBACK on the connection. There appeared to be legacy encoding issues
+ which prevented the use of "SHOW TABLES", for this, but as MySQL support is
+ now at 5.0.2 or above due to :ticket:`4189`, the information_schema tables
+ are now available in all cases.
+
.. sourcecode:: pycon+sql
{sql}>>> metadata.create_all(engine)
- PRAGMA...
+ BEGIN...
CREATE TABLE users (
id INTEGER NOT NULL,
name VARCHAR,
fullname VARCHAR,
PRIMARY KEY (id)
)
+ <BLANKLINE>
+ <BLANKLINE>
[...] ()
- COMMIT
+ <BLANKLINE>
CREATE TABLE addresses (
id INTEGER NOT NULL,
user_id INTEGER,
PRIMARY KEY (id),
FOREIGN KEY(user_id) REFERENCES users (id)
)
+ <BLANKLINE>
+ <BLANKLINE>
[...] ()
COMMIT
unambiguous, and incremental upgrade path in order to migrate applications to
being fully 2.0 compatible. The :class:`.exc.RemovedIn20Warning` deprecation
warning is at the base of this system to provide guidance on what behaviors in
-an existing codebase will need to be modified.
-
-For some occurrences of this warning, an additional recommendation to use an
-API in either the ``sqlalchemy.future`` or ``sqlalchemy.future.orm`` packages
-may be present. This refers to two special future-compatibility packages that
-are part of SQLAlchemy 1.4 and are there to help migrate an application to the
-2.0 version.
+an existing codebase will need to be modified. An overview of how to enable
+this warning is at :ref:`deprecation_20_mode`.
.. seealso::
the 1.x series, as well as the current goals and progress of SQLAlchemy
2.0.
+
+ :ref:`deprecation_20_mode` - specific guidelines on how to use
+ "2.0 deprecations mode" in SQLAlchemy 1.4.
+
.. _error_c9bf:
A bind was located via legacy bound metadata, but since future=True is set on this Session, this bind is ignored.
.. sourcecode:: python+sql
>>> Base.metadata.create_all(engine)
- PRAGMA main.table_info("users")
- [...] ()
- PRAGMA temp.table_info("users")
- [...] ()
+ BEGIN...
CREATE TABLE users (
- id INTEGER NOT NULL, name VARCHAR,
+ id INTEGER NOT NULL,
+ name VARCHAR,
fullname VARCHAR,
nickname VARCHAR,
PRIMARY KEY (id)
)
+ <BLANKLINE>
+ <BLANKLINE>
[...] ()
COMMIT
.. sourcecode:: python+sql
{sql}>>> Base.metadata.create_all(engine)
- PRAGMA...
+ BEGIN...
CREATE TABLE addresses (
id INTEGER NOT NULL,
email_address VARCHAR NOT NULL,
user_id INTEGER,
PRIMARY KEY (id),
- FOREIGN KEY(user_id) REFERENCES users (id)
+ FOREIGN KEY(user_id) REFERENCES users (id)
)
+ <BLANKLINE>
+ <BLANKLINE>
[...] ()
COMMIT
.. sourcecode:: python+sql
{sql}>>> Base.metadata.create_all(engine)
- PRAGMA...
+ BEGIN...
CREATE TABLE keywords (
id INTEGER NOT NULL,
keyword VARCHAR(50) NOT NULL,
PRIMARY KEY (id),
UNIQUE (keyword)
)
+ <BLANKLINE>
+ <BLANKLINE>
[...] ()
- COMMIT
+ <BLANKLINE>
CREATE TABLE posts (
id INTEGER NOT NULL,
user_id INTEGER,
PRIMARY KEY (id),
FOREIGN KEY(user_id) REFERENCES users (id)
)
+ <BLANKLINE>
+ <BLANKLINE>
[...] ()
- COMMIT
+ <BLANKLINE>
CREATE TABLE post_keywords (
post_id INTEGER NOT NULL,
keyword_id INTEGER NOT NULL,
FOREIGN KEY(post_id) REFERENCES posts (id),
FOREIGN KEY(keyword_id) REFERENCES keywords (id)
)
+ <BLANKLINE>
+ <BLANKLINE>
[...] ()
COMMIT
import re
from sqlalchemy import literal_column
+from sqlalchemy import text
from sqlalchemy.sql import visitors
from . import reflection as _reflection
from .enumerated import ENUM
from ...sql import elements
from ...sql import roles
from ...sql import util as sql_util
+from ...sql.sqltypes import Unicode
from ...types import BINARY
from ...types import BLOB
from ...types import BOOLEAN
return connection.exec_driver_sql("SELECT DATABASE()").scalar()
def has_table(self, connection, table_name, schema=None):
- # SHOW TABLE STATUS LIKE and SHOW TABLES LIKE do not function properly
- # on macosx (and maybe win?) with multibyte table names.
- #
- # TODO: if this is not a problem on win, make the strategy swappable
- # based on platform. DESCRIBE is slower.
-
- # [ticket:726]
- # full_name = self.identifier_preparer.format_table(table,
- # use_schema=True)
+ if schema is None:
+ schema = self.default_schema_name
- full_name = ".".join(
- self.identifier_preparer._quote_free_identifiers(
- schema, table_name
- )
+ rs = connection.execute(
+ text(
+ "SELECT * FROM information_schema.tables WHERE "
+ "table_schema = :table_schema AND "
+ "table_name = :table_name"
+ ).bindparams(
+ sql.bindparam("table_schema", type_=Unicode),
+ sql.bindparam("table_name", type_=Unicode),
+ ),
+ {
+ "table_schema": util.text_type(schema),
+ "table_name": util.text_type(table_name),
+ },
)
-
- st = "DESCRIBE %s" % full_name
- rs = None
- try:
- try:
- rs = connection.execution_options(
- skip_user_error_events=True
- ).exec_driver_sql(st)
- have = rs.fetchone() is not None
- rs.close()
- return have
- except exc.DBAPIError as e:
- if self._extract_error_code(e.orig) == 1146:
- return False
- raise
- finally:
- if rs:
- rs.close()
+ return bool(rs.scalar())
def has_sequence(self, connection, sequence_name, schema=None):
if not self.supports_sequences:
def _commit_impl(self, autocommit=False):
assert not self.__branch_from
+ if autocommit:
+ util.warn_deprecated_20(
+ "The current statement is being autocommitted using "
+ "implicit autocommit, which will be removed in "
+ "SQLAlchemy 2.0. "
+ "Use the .begin() method of Engine or Connection in order to "
+ "use an explicit transaction for DML and DDL statements."
+ )
+
if self._has_events or self.engine._has_events:
self.dispatch.commit(self)
return conn.run_callable(callable_, *args, **kwargs)
def _run_ddl_visitor(self, visitorcallable, element, **kwargs):
- with self.connect() as conn:
+ with self.begin() as conn:
conn._run_ddl_visitor(visitorcallable, element, **kwargs)
@util.deprecated_20(
:ref:`error_b8d9`.
+ :ref:`deprecation_20_mode`
+
"""
deprecated_since = "1.4"
from .exclusions import only_on # noqa
from .exclusions import skip # noqa
from .exclusions import skip_if # noqa
+from .schema import eq_type_affinity # noqa
from .util import adict # noqa
from .util import fail # noqa
from .util import flag_combinations # noqa
def expect_warnings(*messages, **kw):
"""Context manager which expects one or more warnings.
- With no arguments, squelches all SAWarnings emitted via
+ With no arguments, squelches all SAWarning and RemovedIn20Warning emitted via
sqlalchemy.util.warn and sqlalchemy.util.warn_limited. Otherwise
pass string expressions that will match selected warnings via regex;
all non-matching warnings are sent through.
Note that the test suite sets SAWarning warnings to raise exceptions.
- """
- return _expect_warnings(sa_exc.SAWarning, messages, **kw)
+ """ # noqa
+ return _expect_warnings(
+ (sa_exc.SAWarning, sa_exc.RemovedIn20Warning), messages, **kw
+ )
@contextlib.contextmanager
from . import exclusions
from .. import event
from .. import schema
+from .. import types as sqltypes
__all__ = ["Table", "Column"]
return col
+class eq_type_affinity(object):
+ """Helper to compare types inside of datastructures based on affinity.
+
+ E.g.::
+
+ eq_(
+ inspect(connection).get_columns("foo"),
+ [
+ {
+ "name": "id",
+ "type": testing.eq_type_affinity(sqltypes.INTEGER),
+ "nullable": False,
+ "default": None,
+ "autoincrement": False,
+ },
+ {
+ "name": "data",
+ "type": testing.eq_type_affinity(sqltypes.NullType),
+ "nullable": True,
+ "default": None,
+ "autoincrement": False,
+ },
+ ],
+ )
+
+ """
+
+ def __init__(self, target):
+ self.target = sqltypes.to_instance(target)
+
+ def __eq__(self, other):
+ return self.target._type_affinity is other._type_affinity
+
+ def __ne__(self, other):
+ return self.target._type_affinity is not other._type_affinity
+
+
def _truncate_name(dialect, name):
if len(name) > dialect.max_identifier_length:
return (
t = Table("t", self.metadata, Column("x", type_))
t.create()
- with testing.db.connect() as conn:
+ with testing.db.begin() as conn:
for value in input_:
ins = (
t.insert()
def test_orm(self):
self._run_doctest("orm/tutorial.rst")
- @testing.emits_warning("SELECT statement has a cartesian")
+ @testing.emits_warning()
def test_core(self):
self._run_doctest("core/tutorial.rst")
from sqlalchemy import Table
from sqlalchemy import testing
from sqlalchemy import types
+from sqlalchemy import types as sqltypes
from sqlalchemy import util
from sqlalchemy.dialects import mssql
from sqlalchemy.dialects.mssql import base
eq_(table2.c["col1"].dialect_options["mssql"]["identity_start"], 2)
eq_(table2.c["col1"].dialect_options["mssql"]["identity_increment"], 3)
- @testing.emits_warning("Did not recognize")
@testing.provide_metadata
- def test_skip_types(self):
- metadata = self.metadata
- with testing.db.connect() as c:
- c.exec_driver_sql(
- "create table foo (id integer primary key, data xml)"
- )
+ def test_skip_types(self, connection):
+ connection.exec_driver_sql(
+ "create table foo (id integer primary key, data xml)"
+ )
with mock.patch.object(
- testing.db.dialect, "ischema_names", {"int": mssql.INTEGER}
+ connection.dialect, "ischema_names", {"int": mssql.INTEGER}
):
- t1 = Table("foo", metadata, autoload=True)
- assert isinstance(t1.c.id.type, Integer)
- assert isinstance(t1.c.data.type, types.NullType)
+ with testing.expect_warnings(
+ "Did not recognize type 'xml' of column 'data'"
+ ):
+ eq_(
+ inspect(connection).get_columns("foo"),
+ [
+ {
+ "name": "id",
+ "type": testing.eq_type_affinity(sqltypes.INTEGER),
+ "nullable": False,
+ "default": None,
+ "autoincrement": False,
+ },
+ {
+ "name": "data",
+ "type": testing.eq_type_affinity(
+ sqltypes.NullType
+ ),
+ "nullable": True,
+ "default": None,
+ "autoincrement": False,
+ },
+ ],
+ )
@testing.provide_metadata
def test_cross_schema_fk_pk_name_overlaps(self):
from sqlalchemy import FLOAT
from sqlalchemy import ForeignKey
from sqlalchemy import ForeignKeyConstraint
+from sqlalchemy import func
from sqlalchemy import Index
from sqlalchemy import inspect
from sqlalchemy import INTEGER
from sqlalchemy import Numeric
from sqlalchemy import PrimaryKeyConstraint
from sqlalchemy import select
-from sqlalchemy import String
from sqlalchemy import testing
from sqlalchemy import text
from sqlalchemy import Unicode
eq_(set(t.name for t in m.tables.values()), set(["admin_docindex"]))
-class UnsupportedIndexReflectTest(fixtures.TestBase):
- __only_on__ = "oracle"
- __backend__ = True
-
- @testing.emits_warning("No column names")
- @testing.provide_metadata
- def test_reflect_functional_index(self):
- metadata = self.metadata
- Table(
- "test_index_reflect",
- metadata,
- Column("data", String(20), primary_key=True),
- )
- metadata.create_all()
-
- exec_sql(
- testing.db,
- "CREATE INDEX DATA_IDX ON " "TEST_INDEX_REFLECT (UPPER(DATA))",
- )
- m2 = MetaData(testing.db)
- Table("test_index_reflect", m2, autoload=True)
-
-
def all_tables_compression_missing():
try:
exec_sql(testing.db, "SELECT compression FROM all_tables")
],
)
+ @testing.provide_metadata
+ def test_reflect_fn_index(self, connection):
+ """test reflection of a functional index.
+
+ it appears this emitted a warning at some point but does not right now.
+ the returned data is not exactly correct, but this is what it's
+ likely been doing for many years.
+
+ """
+
+ metadata = self.metadata
+ s_table = Table(
+ "sometable",
+ metadata,
+ Column("group", Unicode(255), primary_key=True),
+ Column("col", Unicode(255)),
+ )
+
+ Index("data_idx", func.upper(s_table.c.col))
+
+ metadata.create_all(connection)
+
+ eq_(
+ inspect(connection).get_indexes("sometable"),
+ [
+ {
+ "column_names": [],
+ "dialect_options": {},
+ "name": "data_idx",
+ "unique": False,
+ }
+ ],
+ )
+
@testing.provide_metadata
def test_basic(self):
metadata = self.metadata
)
+class DDLTransactionTest(fixtures.TestBase):
+ """test DDL transactional behavior as of SQLAlchemy 1.4."""
+
+ @testing.fixture
+ def metadata_fixture(self):
+ m = MetaData()
+ Table("t1", m, Column("q", Integer))
+ Table("t2", m, Column("q", Integer))
+
+ try:
+ yield m
+ finally:
+ m.drop_all(testing.db)
+
+ def _listening_engine_fixture(self, future=False):
+ eng = engines.testing_engine(future=future)
+
+ m1 = mock.Mock()
+
+ event.listen(eng, "begin", m1.begin)
+ event.listen(eng, "commit", m1.commit)
+ event.listen(eng, "rollback", m1.rollback)
+
+ @event.listens_for(eng, "before_cursor_execute")
+ def before_cursor_execute(
+ conn, cursor, statement, parameters, context, executemany
+ ):
+ if "CREATE TABLE" in statement:
+ m1.cursor_execute("CREATE TABLE ...")
+
+ eng.connect().close()
+
+ return eng, m1
+
+ @testing.fixture
+ def listening_engine_fixture(self):
+ return self._listening_engine_fixture(future=False)
+
+ @testing.fixture
+ def future_listening_engine_fixture(self):
+ return self._listening_engine_fixture(future=True)
+
+ def test_ddl_legacy_engine(
+ self, metadata_fixture, listening_engine_fixture
+ ):
+ eng, m1 = listening_engine_fixture
+
+ metadata_fixture.create_all(eng)
+
+ eq_(
+ m1.mock_calls,
+ [
+ mock.call.begin(mock.ANY),
+ mock.call.cursor_execute("CREATE TABLE ..."),
+ mock.call.cursor_execute("CREATE TABLE ..."),
+ mock.call.commit(mock.ANY),
+ ],
+ )
+
+ def test_ddl_future_engine(
+ self, metadata_fixture, future_listening_engine_fixture
+ ):
+ eng, m1 = future_listening_engine_fixture
+
+ metadata_fixture.create_all(eng)
+
+ eq_(
+ m1.mock_calls,
+ [
+ mock.call.begin(mock.ANY),
+ mock.call.cursor_execute("CREATE TABLE ..."),
+ mock.call.cursor_execute("CREATE TABLE ..."),
+ mock.call.commit(mock.ANY),
+ ],
+ )
+
+ def test_ddl_legacy_connection_no_transaction(
+ self, metadata_fixture, listening_engine_fixture
+ ):
+ eng, m1 = listening_engine_fixture
+
+ with eng.connect() as conn:
+ with testing.expect_deprecated(
+ "The current statement is being autocommitted using "
+ "implicit autocommit"
+ ):
+ metadata_fixture.create_all(conn)
+
+ eq_(
+ m1.mock_calls,
+ [
+ mock.call.cursor_execute("CREATE TABLE ..."),
+ mock.call.commit(mock.ANY),
+ mock.call.cursor_execute("CREATE TABLE ..."),
+ mock.call.commit(mock.ANY),
+ ],
+ )
+
+ def test_ddl_legacy_connection_transaction(
+ self, metadata_fixture, listening_engine_fixture
+ ):
+ eng, m1 = listening_engine_fixture
+
+ with eng.connect() as conn:
+ with conn.begin():
+ metadata_fixture.create_all(conn)
+
+ eq_(
+ m1.mock_calls,
+ [
+ mock.call.begin(mock.ANY),
+ mock.call.cursor_execute("CREATE TABLE ..."),
+ mock.call.cursor_execute("CREATE TABLE ..."),
+ mock.call.commit(mock.ANY),
+ ],
+ )
+
+ def test_ddl_future_connection_autobegin_transaction(
+ self, metadata_fixture, future_listening_engine_fixture
+ ):
+ eng, m1 = future_listening_engine_fixture
+
+ with eng.connect() as conn:
+ metadata_fixture.create_all(conn)
+
+ conn.commit()
+
+ eq_(
+ m1.mock_calls,
+ [
+ mock.call.begin(mock.ANY),
+ mock.call.cursor_execute("CREATE TABLE ..."),
+ mock.call.cursor_execute("CREATE TABLE ..."),
+ mock.call.commit(mock.ANY),
+ ],
+ )
+
+ def test_ddl_future_connection_explicit_begin_transaction(
+ self, metadata_fixture, future_listening_engine_fixture
+ ):
+ eng, m1 = future_listening_engine_fixture
+
+ with eng.connect() as conn:
+ with conn.begin():
+ metadata_fixture.create_all(conn)
+
+ eq_(
+ m1.mock_calls,
+ [
+ mock.call.begin(mock.ANY),
+ mock.call.cursor_execute("CREATE TABLE ..."),
+ mock.call.cursor_execute("CREATE TABLE ..."),
+ mock.call.commit(mock.ANY),
+ ],
+ )
+
+
class DDLTest(fixtures.TestBase, AssertsCompiledSQL):
def mock_engine(self):
def executor(*a, **kw):
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import testing
+from sqlalchemy import text
from sqlalchemy import VARCHAR
from sqlalchemy.engine import reflection
from sqlalchemy.engine.base import Connection
)
-class TransactionTest(fixtures.TestBase):
+class TransactionTest(fixtures.TablesTest):
__backend__ = True
@classmethod
- def setup_class(cls):
- metadata = MetaData()
- cls.users = Table(
- "query_users",
+ def define_tables(cls, metadata):
+ Table(
+ "users",
metadata,
Column("user_id", Integer, primary_key=True),
Column("user_name", String(20)),
test_needs_acid=True,
)
- cls.users.create(testing.db)
-
- def teardown(self):
- with testing.db.connect() as conn:
- conn.execute(self.users.delete())
-
- @classmethod
- def teardown_class(cls):
- cls.users.drop(testing.db)
+ Table("inserttable", metadata, Column("data", String(20)))
def test_transaction_container(self):
- users = self.users
+ users = self.tables.users
def go(conn, table, data):
for d in data:
with testing.db.connect() as conn:
eq_(conn.execute(users.select()).fetchall(), [(1, "user1")])
+ def test_implicit_autocommit_compiled(self):
+ users = self.tables.users
+
+ with testing.db.connect() as conn:
+ with testing.expect_deprecated_20(
+ "The current statement is being autocommitted "
+ "using implicit autocommit."
+ ):
+ conn.execute(
+ users.insert(), {"user_id": 1, "user_name": "user3"}
+ )
+
+ def test_implicit_autocommit_text(self):
+ with testing.db.connect() as conn:
+ with testing.expect_deprecated_20(
+ "The current statement is being autocommitted "
+ "using implicit autocommit."
+ ):
+ conn.execute(
+ text("insert into inserttable (data) values ('thedata')")
+ )
+
+ def test_implicit_autocommit_driversql(self):
+ with testing.db.connect() as conn:
+ with testing.expect_deprecated_20(
+ "The current statement is being autocommitted "
+ "using implicit autocommit."
+ ):
+ conn.exec_driver_sql(
+ "insert into inserttable (data) values ('thedata')"
+ )
+
class HandleInvalidatedOnConnectTest(fixtures.TestBase):
__requires__ = ("sqlite",)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
