--- /dev/null
+.. change::
+ :tags: change, engine
+ :tickets: 9647
+
+ An empty sequence passed to any ``execute()`` method now
+ raised a deprecation warning, since such an executemany
+ is invalid.
+ Pull request courtesy of Carlos Sousa.
import operator
from typing import Any
from typing import Optional
-from typing import Sequence
from typing import Tuple
from typing import TYPE_CHECKING
-from sqlalchemy import exc
+from .. import exc
+from ..util import warn_deprecated
if TYPE_CHECKING:
from .interfaces import _CoreAnyExecuteParams
@cython.inline
@cython.cfunc
-def _is_mapping_or_tuple(value: object) -> cython.bint:
+def _is_mapping_or_tuple(value: object, /) -> cython.bint:
return (
isinstance(value, dict)
or isinstance(value, tuple)
)
-@cython.inline
-@cython.cfunc
-@cython.exceptval(0)
-def _validate_execute_many_item(params: Sequence[Any]) -> cython.bint:
- ret: cython.bint = 1
- if len(params) > 0:
- if not _is_mapping_or_tuple(params[0]):
- ret = 0
- raise exc.ArgumentError(
- "List argument must consist only of tuples or dictionaries"
- )
- return ret
-
-
-# _is_mapping_or_tuple and _validate_execute_many_item could be
-# inlined if pure python perf is a problem
+# _is_mapping_or_tuple could be inlined if pure python perf is a problem
def _distill_params_20(
params: Optional[_CoreAnyExecuteParams],
) -> _CoreMultiExecuteParams:
# Assume list is more likely than tuple
elif isinstance(params, list) or isinstance(params, tuple):
# collections_abc.MutableSequence # avoid abc.__instancecheck__
- _validate_execute_many_item(params)
+ if len(params) == 0:
+ warn_deprecated(
+ "Empty parameter sequence passed to execute(). "
+ "This use is deprecated and will raise an exception in a "
+ "future SQLAlchemy release",
+ "2.1",
+ )
+ elif not _is_mapping_or_tuple(params[0]):
+ raise exc.ArgumentError(
+ "List argument must consist only of tuples or dictionaries"
+ )
return params
elif isinstance(params, dict) or isinstance(params, Mapping):
# only do immutabledict or abc.__instancecheck__ for Mapping after
return _Empty_Tuple
elif isinstance(params, list):
# collections_abc.MutableSequence # avoid abc.__instancecheck__
- _validate_execute_many_item(params)
+ if len(params) > 0 and not _is_mapping_or_tuple(params[0]):
+ raise exc.ArgumentError(
+ "List argument must consist only of tuples or dictionaries"
+ )
return params
elif _is_mapping_or_tuple(params):
return [params] # type: ignore[return-value]
)
else:
result = conn.execute(
- statement, params or {}, execution_options=execution_options
+ statement, params, execution_options=execution_options
)
if _scalar_result:
],
)
- def _assert_result(self, conn, select, result, params=()):
+ def _assert_result(self, conn, select, result, params=None):
eq_(conn.execute(select, params).fetchall(), result)
def test_plain_union(self, connection):
)
def _assert_result(
- self, connection, select, result, params=(), set_=False
+ self, connection, select, result, params=None, set_=False
):
if set_:
query_res = connection.execute(select, params).fetchall()
else:
eq_(connection.execute(select, params).fetchall(), result)
- def _assert_result_str(self, select, result, params=()):
+ def _assert_result_str(self, select, result, params=None):
with config.db.connect() as conn:
eq_(conn.exec_driver_sql(select, params).fetchall(), result)
class JoinTest(fixtures.TablesTest):
__backend__ = True
- def _assert_result(self, select, result, params=()):
+ def _assert_result(self, select, result, params=None):
with config.db.connect() as conn:
eq_(conn.execute(select, params).fetchall(), result)
],
)
- def _assert_result(self, select, result, params=()):
+ def _assert_result(self, select, result, params=None):
with config.db.connect() as conn:
eq_(conn.execute(select, params).fetchall(), result)
],
)
- def _assert_result(self, select, result, params=()):
+ def _assert_result(self, select, result, params=None):
with config.db.connect() as conn:
eq_(conn.execute(select, params).fetchall(), result)
from sqlalchemy.dialects.oracle import oracledb
from sqlalchemy.sql import column
from sqlalchemy.sql.sqltypes import NullType
-from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import AssertsCompiledSQL
from sqlalchemy.testing import eq_
from sqlalchemy.testing import expect_raises_message
)
eq_(actual, self.data)
- # this comes from cx_Oracle because these are raw
- # cx_Oracle.Variable objects
- if testing.requires.oracle5x.enabled:
- assert_raises_message(
- testing.db.dialect.dbapi.ProgrammingError,
- "LOB variable no longer valid after subsequent fetch",
- go,
- )
- else:
- go()
+ go()
def test_lobs_with_convert_many_rows(self):
# even with low arraysize, lobs are fine in autoconvert
from sqlalchemy.testing import is_false
from sqlalchemy.testing import is_not
from sqlalchemy.testing import is_true
+from sqlalchemy.testing.assertions import expect_deprecated
from sqlalchemy.testing.assertsql import CompiledSQL
from sqlalchemy.testing.provision import normalize_sequence
from sqlalchemy.testing.schema import Column
conn.close()
def test_empty_insert(self, connection):
- """test that execute() interprets [] as a list with no params"""
+ """test that execute() interprets [] as a list with no params and
+ warns since it has nothing to do with such an executemany.
+ """
users_autoinc = self.tables.users_autoinc
- connection.execute(
- users_autoinc.insert().values(user_name=bindparam("name", None)),
- [],
- )
- eq_(connection.execute(users_autoinc.select()).fetchall(), [(1, None)])
+ with expect_deprecated(
+ r"Empty parameter sequence passed to execute\(\). "
+ "This use is deprecated and will raise an exception in a "
+ "future SQLAlchemy release"
+ ):
+ connection.execute(
+ users_autoinc.insert().values(
+ user_name=bindparam("name", None)
+ ),
+ [],
+ )
+
+ eq_(len(connection.execute(users_autoinc.select()).all()), 1)
+
+ @testing.only_on("sqlite")
+ def test_raw_insert_with_empty_list(self, connection):
+ """exec_driver_sql instead does not raise if an empty list is passed.
+ Let the driver do that if it wants to.
+ """
+ conn = connection
+ with expect_raises_message(
+ tsa.exc.ProgrammingError, "Incorrect number of bindings supplied"
+ ):
+ conn.exec_driver_sql(
+ "insert into users (user_id, user_name) values (?, ?)", []
+ )
@testing.only_on("sqlite")
def test_execute_compiled_favors_compiled_paramstyle(self):
from sqlalchemy.testing import expect_raises_message
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_none
+from sqlalchemy.testing.assertions import expect_deprecated
from sqlalchemy.util import immutabledict
eq_(self.module._distill_params_20(None), ())
def test_distill_20_empty_sequence(self):
- eq_(self.module._distill_params_20(()), ())
- eq_(self.module._distill_params_20([]), [])
+ with expect_deprecated(
+ r"Empty parameter sequence passed to execute\(\). "
+ "This use is deprecated and will raise an exception in a "
+ "future SQLAlchemy release"
+ ):
+ eq_(self.module._distill_params_20(()), ())
+ eq_(self.module._distill_params_20([]), [])
def test_distill_20_sequence_sequence(self):
eq_(self.module._distill_params_20(((1, 2, 3),)), ((1, 2, 3),))
):
sess.scalar("select id from users where id=:id", {"id": 7})
+ @testing.skip_if(
+ "oracle", "missing SELECT keyword [SQL: INSERT INTO tbl () VALUES ()]"
+ )
+ def test_empty_list_execute(self, metadata, connection):
+ t = Table("tbl", metadata, Column("col", sa.Integer))
+ t.create(connection)
+ sess = Session(bind=connection)
+ sess.execute(t.insert(), {"col": 42})
+
+ with assertions.expect_deprecated(
+ r"Empty parameter sequence passed to execute\(\). "
+ "This use is deprecated and will raise an exception in a "
+ "future SQLAlchemy release"
+ ):
+ sess.execute(t.insert(), [])
+
+ eq_(len(sess.execute(sa.select(t.c.col)).all()), 2)
+
class TransScopingTest(_fixtures.FixtureTest):
run_inserts = None
def none_20(self):
self.impl._distill_params_20(None)
- @test_case
- def empty_sequence_20(self):
- self.impl._distill_params_20(())
- self.impl._distill_params_20([])
-
@test_case
def list_20(self):
self.impl._distill_params_20(self.list_tup)
assert not py_util._is_compiled()
return py_util.tuplegetter
- @staticmethod
- def c():
- from sqlalchemy import cresultproxy
-
- return cresultproxy.tuplegetter
-
@staticmethod
def cython():
from sqlalchemy.engine import _util_cy
IMPLEMENTATIONS = {
"python": python.__func__,
- "c": c.__func__,
"cython": cython.__func__,
}
@property
def returning_star(self):
- """backend supports RETURNING *"""
+ """backend supports ``RETURNING *``"""
return skip_if(["oracle", "mssql"])
return only_if(go)
- @property
- def oracle5x(self):
- return only_if(
- lambda config: against(config, "oracle+cx_oracle")
- and config.db.dialect.cx_oracle_ver < (6,)
- )
-
@property
def fail_on_oracledb_thin(self):
def go(config):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
