--- /dev/null
+.. change::
+ :tags: engine, removed
+
+ Removed the previously deprecated ``case_sensitive`` parameter from
+ :func:`_sa.create_engine`, which would impact only the lookup of string
+ column names in Core-only result set rows; it had no effect on the behavior
+ of the ORM. The effective behavior of what ``case_sensitive`` refers
+ towards remains at its default value of ``True``, meaning that string names
+ looked up in ``row._mapping`` will match case-sensitively, just like any
+ other Python mapping.
+
+ Note that the ``case_sensitive`` parameter was not in any way related to
+ the general subject of case sensitivity control, quoting, and "name
+ normalization" (i.e. converting for databases that consider all uppercase
+ words to be case insensitive) for DDL identifier names, which remains a
+ normal core feature of SQLAlchemy.
+
+
'expressions, or an "empty set" SELECT, at statement execution'
"time.",
),
- case_sensitive=(
- "1.4",
- "The :paramref:`_sa.create_engine.case_sensitive` parameter "
- "is deprecated and will be removed in a future release. "
- "Applications should work with result column names in a case "
- "sensitive fashion.",
- ),
)
def create_engine(url, **kwargs):
"""Create a new :class:`_engine.Engine` instance.
:ref:`connections_toplevel`
- :param case_sensitive: if False, result column names
- will match in a case-insensitive fashion, that is,
- ``row['SomeColumn']``.
-
:param connect_args: a dictionary of options which will be
passed directly to the DBAPI's ``connect()`` method as
additional keyword arguments. See the example
__slots__ = (
"_keymap",
- "case_sensitive",
"_processors",
"_keys",
"_keymap_by_result_column_idx",
tup = tuplegetter(*indexes)
new_metadata = self.__class__.__new__(self.__class__)
- new_metadata.case_sensitive = self.case_sensitive
new_metadata._unpickled = self._unpickled
new_metadata._processors = self._processors
new_metadata._keys = new_keys
else:
md._keymap[new] = rec
- md.case_sensitive = self.case_sensitive
md._unpickled = self._unpickled
md._processors = self._processors
assert not self._tuplefilter
def __init__(self, parent, cursor_description):
context = parent.context
- dialect = context.dialect
self._tuplefilter = None
self._translated_indexes = None
- self.case_sensitive = dialect.case_sensitive
self._safe_for_cache = self._unpickled = False
if context.result_column_struct:
for key in (metadata_entry[MD_RENDERED_NAME],) + (
metadata_entry[MD_OBJECTS] or ()
):
- if not self.case_sensitive and isinstance(
- key, util.string_types
- ):
- key = key.lower()
idx = metadata_entry[MD_INDEX]
# if this key has been associated with more than one
# positional index, it's a dupe
"""
- case_sensitive = context.dialect.case_sensitive
-
if (
num_ctx_cols
and cols_are_ordered
idx,
idx,
rmap_entry[RM_OBJECTS],
- rmap_entry[RM_NAME].lower()
- if not case_sensitive
- else rmap_entry[RM_NAME],
+ rmap_entry[RM_NAME],
rmap_entry[RM_RENDERED_NAME],
context.get_result_processor(
rmap_entry[RM_TYPE],
"""
dialect = context.dialect
- case_sensitive = dialect.case_sensitive
translate_colname = context._translate_colname
description_decoder = (
dialect._description_decoder
colname = normalize_name(colname)
self._keys.append(colname)
- if not case_sensitive:
- colname = colname.lower()
yield idx, colname, untranslated, coltype
result_columns,
loose_column_name_matching,
):
- dialect = context.dialect
- case_sensitive = dialect.case_sensitive
match_map = self._create_description_match_map(
- result_columns, case_sensitive, loose_column_name_matching
+ result_columns, loose_column_name_matching
)
for (
idx,
def _create_description_match_map(
cls,
result_columns,
- case_sensitive=True,
loose_column_name_matching=False,
):
"""when matching cursor.description to a set of names that are present
for ridx, elem in enumerate(result_columns):
key = elem[RM_RENDERED_NAME]
- if not case_sensitive:
- key = key.lower()
if key in d:
# conflicting keyname - just add the column-linked objects
# to the existing record. if there is a duplicate column
def _key_fallback(self, key, err, raiseerr=True):
- # we apparently have not marked .case_sensitive as
- # RemovedIn20. I still think we should remove it as I can't
- # imagine anyone is using it, however lets make that a separate
- # commit.
- if not self.case_sensitive and isinstance(key, util.string_types):
- map_ = self._keymap
- result = map_.get(key.lower())
- if result is not None:
- return result
-
if raiseerr:
if self._unpickled and isinstance(key, elements.ColumnElement):
util.raise_(
if isinstance(key, util.string_types + util.int_types)
},
"_keys": self._keys,
- "case_sensitive": self.case_sensitive,
"_translated_indexes": self._translated_indexes,
"_tuplefilter": self._tuplefilter,
}
rec[MD_RESULT_MAP_INDEX]: rec for rec in self._keymap.values()
}
self._keys = state["_keys"]
- self.case_sensitive = state["case_sensitive"]
self._unpickled = True
if state["_translated_indexes"]:
self._translated_indexes = state["_translated_indexes"]
'expressions, or an "empty set" SELECT, at statement execution'
"time.",
),
- case_sensitive=(
- "1.4",
- "The :paramref:`_sa.create_engine.case_sensitive` parameter "
- "is deprecated and will be removed in a future release. "
- "Applications should work with result column names in a case "
- "sensitive fashion.",
- ),
server_side_cursors=(
"1.4",
"The :paramref:`_sa.create_engine.server_side_cursors` parameter "
paramstyle=None,
dbapi=None,
implicit_returning=None,
- case_sensitive=True,
supports_native_boolean=None,
max_identifier_length=None,
label_length=None,
self.type_compiler = self.type_compiler(self)
if supports_native_boolean is not None:
self.supports_native_boolean = supports_native_boolean
- self.case_sensitive = case_sensitive
self._user_defined_max_identifier_length = max_identifier_length
if self._user_defined_max_identifier_length:
from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy import inspect
-from sqlalchemy import INT
from sqlalchemy import Integer
from sqlalchemy import join
from sqlalchemy import literal_column
from sqlalchemy import testing
from sqlalchemy import text
from sqlalchemy import util
-from sqlalchemy import VARCHAR
from sqlalchemy.engine import default
from sqlalchemy.sql import coercions
from sqlalchemy.sql import LABEL_STYLE_TABLENAME_PLUS_COL
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import assertions
from sqlalchemy.testing import AssertsCompiledSQL
-from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
-from sqlalchemy.testing import in_
from sqlalchemy.testing import is_
from sqlalchemy.testing import is_false
from sqlalchemy.testing import is_true
self._test_autoincrement(conn)
-class CursorResultTest(fixtures.TablesTest):
- __backend__ = True
-
- @classmethod
- def define_tables(cls, metadata):
- Table(
- "users",
- metadata,
- Column(
- "user_id", INT, primary_key=True, test_needs_autoincrement=True
- ),
- Column("user_name", VARCHAR(20)),
- test_needs_acid=True,
- )
- Table(
- "addresses",
- metadata,
- Column(
- "address_id",
- Integer,
- primary_key=True,
- test_needs_autoincrement=True,
- ),
- Column("user_id", Integer, ForeignKey("users.user_id")),
- Column("address", String(30)),
- test_needs_acid=True,
- )
-
- Table(
- "users2",
- metadata,
- Column("user_id", INT, primary_key=True),
- Column("user_name", VARCHAR(20)),
- test_needs_acid=True,
- )
-
- @classmethod
- def insert_data(cls, connection):
- users = cls.tables.users
-
- connection.execute(
- users.insert(),
- [
- dict(user_id=1, user_name="john"),
- dict(user_id=2, user_name="jack"),
- ],
- )
-
- @testing.requires.duplicate_names_in_cursor_description
- def test_ambiguous_column_case_sensitive(self):
- with testing.expect_deprecated(
- "The create_engine.case_sensitive parameter is deprecated"
- ):
- eng = engines.testing_engine(options=dict(case_sensitive=False))
-
- with eng.connect() as conn:
- row = conn.execute(
- select(
- literal_column("1").label("SOMECOL"),
- literal_column("1").label("SOMECOL"),
- )
- ).first()
-
- assert_raises_message(
- exc.InvalidRequestError,
- "Ambiguous column name",
- lambda: row._mapping["somecol"],
- )
-
- def test_row_case_insensitive(self):
- with testing.expect_deprecated(
- "The create_engine.case_sensitive parameter is deprecated"
- ):
- with engines.testing_engine(
- options={"case_sensitive": False}
- ).connect() as ins_conn:
- row = ins_conn.execute(
- select(
- literal_column("1").label("case_insensitive"),
- literal_column("2").label("CaseSensitive"),
- )
- ).first()
-
- eq_(
- list(row._mapping.keys()),
- ["case_insensitive", "CaseSensitive"],
- )
-
- in_("case_insensitive", row._keymap)
- in_("CaseSensitive", row._keymap)
- in_("casesensitive", row._keymap)
-
- eq_(row._mapping["case_insensitive"], 1)
- eq_(row._mapping["CaseSensitive"], 2)
- eq_(row._mapping["Case_insensitive"], 1)
- eq_(row._mapping["casesensitive"], 2)
-
- def test_row_case_insensitive_unoptimized(self):
- with testing.expect_deprecated(
- "The create_engine.case_sensitive parameter is deprecated"
- ):
- with engines.testing_engine(
- options={"case_sensitive": False}
- ).connect() as ins_conn:
- row = ins_conn.execute(
- select(
- literal_column("1").label("case_insensitive"),
- literal_column("2").label("CaseSensitive"),
- text("3 AS screw_up_the_cols"),
- )
- ).first()
-
- eq_(
- list(row._mapping.keys()),
- ["case_insensitive", "CaseSensitive", "screw_up_the_cols"],
- )
-
- in_("case_insensitive", row._keymap)
- in_("CaseSensitive", row._keymap)
- in_("casesensitive", row._keymap)
-
- eq_(row._mapping["case_insensitive"], 1)
- eq_(row._mapping["CaseSensitive"], 2)
- eq_(row._mapping["screw_up_the_cols"], 3)
- eq_(row._mapping["Case_insensitive"], 1)
- eq_(row._mapping["casesensitive"], 2)
- eq_(row._mapping["screw_UP_the_cols"], 3)
-
-
class DefaultTest(fixtures.TestBase):
__backend__ = True
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
