from __future__ import annotations
+import collections
import decimal
import json
import random
return process
else:
- # for JSON, json decoder is set as an outputtypehandler
+ # not BLOB; the value is decoded by an outputtypehandler
+ # rather than here. see _cx_oracle_outputtypehandler() below
return None
+ def _cx_oracle_outputtypehandler(self, dialect):
+ """Establish an outputtypehandler for JSON result columns.
+
+ The dialect makes use of two distinct outputtypehandlers. One is
+ connection-wide, set up by the dialect's
+ _generate_connection_outputtype_handler(); it sees every column of
+ every statement and knows nothing of SQLAlchemy-side types, only of
+ the DBAPI type the database reports. The other is the one returned
+ here, which the execution context's
+ _generate_cursor_outputtype_handler() installs on the cursor for one
+ particular statement, and which applies only to those result columns
+ that SQLAlchemy knows to be of JSON type.
+
+ The two do not compose. The driver consults the cursor's handler if
+ one is present, and the connection's otherwise, never both. The
+ cursor-level handler assembled by the execution context therefore
+ delegates to the connection-level handler explicitly for the columns
+ it has no handler of its own for, and the DB_TYPE_JSON case below
+ has to repeat what the connection-level handler does rather than
+ defer to it.
+
+ """
+
+ if self._should_use_blob(dialect):
+ # BLOB is decoded by result_processor() instead
+ return None
+
+ cx_Oracle = dialect.dbapi
+ json_deserializer = dialect._json_deserializer
+
+ def handler(cursor, name, default_type, size, precision, scale):
+ if default_type is cx_Oracle.DB_TYPE_JSON:
+ # a native JSON column. the driver decodes these into
+ # Python objects on its own, so a handler is only needed in
+ # order to route the value through a user-supplied
+ # deserializer, receiving it as text to do so. this repeats
+ # the connection-level handler, as noted above
+ if json_deserializer is not None:
+ return cursor.var(
+ cx_Oracle.DB_TYPE_VARCHAR,
+ _CX_ORACLE_MAX_JSON_CONVERTED,
+ cursor.arraysize,
+ outconverter=json_deserializer,
+ )
+ else:
+ return None
+ elif default_type in (
+ cx_Oracle.DB_TYPE_VARCHAR,
+ cx_Oracle.DB_TYPE_NVARCHAR,
+ cx_Oracle.DB_TYPE_LONG,
+ cx_Oracle.DB_TYPE_LONG_NVARCHAR,
+ ):
+ # a JSON-typed expression that isn't a native JSON column,
+ # such as a bound parameter or a string expression selected
+ # directly. the database reports it as ordinary character
+ # data and won't decode it, so decode it here. the
+ # connection-level handler can't do this as it has no way to
+ # know the expression was intended as JSON
+ return cursor.var(
+ default_type,
+ size,
+ cursor.arraysize,
+ outconverter=json_deserializer or json.loads,
+ **dialect._cursor_var_unicode_kwargs,
+ )
+ elif default_type in (cx_Oracle.CLOB, cx_Oracle.NCLOB):
+ # same as the character case above, except the database
+ # reports the expression as a LOB; read it as a string
+ # first, then decode
+ return dialect._cursor_lob_as_str_var(
+ cursor, default_type, json_deserializer or json.loads
+ )
+ else:
+ return None
+
+ return handler
+
class _OracleInteger(sqltypes.Integer):
def get_dbapi_type(self, dbapi):
)
def _generate_cursor_outputtype_handler(self):
- output_handlers = {}
+ assert isinstance(self.compiled, OracleCompiler)
- for keyname, name, objects, type_ in self.compiled._result_columns:
- handler = type_._cached_custom_processor(
+ # accumulate handlers positionally
+ handlers = [
+ type_._cached_custom_processor(
self.dialect,
"cx_oracle_outputtypehandler",
self._get_cx_oracle_type_handler,
)
+ for _, _, _, type_ in self.compiled._result_columns
+ ]
+
+ if not any(handlers):
+ return
+
+ # a handler on the cursor replaces the connection-wide handler
+ # outright rather than adding to it, so keep a reference to the
+ # latter and delegate to it for columns we have no handler for
+ default_handler = self._dbapi_connection.outputtypehandler
- if handler:
- denormalized_name = self.dialect.denormalize_name(keyname)
- output_handlers[denormalized_name] = handler
+ if self.compiled._textual_ordered_columns:
+ # for a textual construct with positional columns, i.e.
+ # text().columns() / tstring().columns(), match handlers
+ # positionally instead of by name, since names are not
+ # deterministic. See #13479
- if output_handlers:
- default_handler = self._dbapi_connection.outputtypehandler
+ handler_queue = collections.deque(handlers)
def output_type_handler(
cursor, name, default_type, size, precision, scale
):
- if name in output_handlers:
- return output_handlers[name](
+ handler = handler_queue.popleft() if handler_queue else None
+ if handler is not None:
+ return handler(
cursor, name, default_type, size, precision, scale
)
else:
cursor, name, default_type, size, precision, scale
)
- self.cursor.outputtypehandler = output_type_handler
+ else:
+ output_handlers = {
+ self.dialect.denormalize_name(rc[0]): handler
+ for rc, handler in zip(self.compiled._result_columns, handlers)
+ if handler
+ }
+
+ def output_type_handler(
+ cursor, name, default_type, size, precision, scale
+ ):
+ handler = output_handlers.get(name)
+ if handler is not None:
+ return handler(
+ cursor, name, default_type, size, precision, scale
+ )
+ else:
+ return default_handler(
+ cursor, name, default_type, size, precision, scale
+ )
+
+ self.cursor.outputtypehandler = output_type_handler
def _get_cx_oracle_type_handler(self, impl):
if hasattr(impl, "_cx_oracle_outputtypehandler"):
_to_decimal = decimal.Decimal
+ def _cursor_lob_as_str_var(self, cursor, default_type, outconverter=None):
+ """Produce a ``cursor.var()`` that receives a CLOB / NCLOB as a
+ string.
+
+ Used both by the connection-wide outputtypehandler and by
+ type-level handlers such as that of :class:`._OracleJson`, which
+ need the LOB contents as a string in order to decode them further.
+
+ """
+ cx_Oracle = self.dbapi
+
+ kw = self._cursor_var_unicode_kwargs
+ if outconverter is not None:
+ kw = {**kw, "outconverter": outconverter}
+
+ return cursor.var(
+ (
+ cx_Oracle.DB_TYPE_VARCHAR
+ if default_type is cx_Oracle.CLOB
+ else cx_Oracle.DB_TYPE_NVARCHAR
+ ),
+ _CX_ORACLE_MAGIC_LOB_SIZE,
+ cursor.arraysize,
+ **kw,
+ )
+
def _generate_connection_outputtype_handler(self):
"""establish the default outputtypehandler established at the
connection level.
+ note that when using a Compiled statement that has types (e.g.
+ TypeEngine), we set up a per-cursor handler instead which supercedes
+ this one, using Oracle-specific TypeEngine handlers delivered by the
+ _cx_oracle_outputtypehandler() method of each one.
+
"""
dialect = self
cx_Oracle.CLOB,
cx_Oracle.NCLOB,
):
- typ = (
- cx_Oracle.DB_TYPE_VARCHAR
- if default_type is cx_Oracle.CLOB
- else cx_Oracle.DB_TYPE_NVARCHAR
- )
- return cursor.var(
- typ,
- _CX_ORACLE_MAGIC_LOB_SIZE,
- cursor.arraysize,
- **dialect._cursor_var_unicode_kwargs,
- )
+ return dialect._cursor_lob_as_str_var(cursor, default_type)
elif dialect.auto_convert_lobs and default_type in (
cx_Oracle.BLOB,
import datetime
import decimal
import functools
+import json
import os
import random
__only_on__ = "oracle+oracledb"
datatype = functools.partial(oracle.JSON, use_blob=True)
+
+
+class TextualSelectTypeHandlerTest(fixtures.TablesTest):
+ """test that per-cursor outputtypehandlers are established for
+ textual constructs that carry positional column information, such as
+ :func:`_sql.text` / :func:`_sql.tstring` with ``.columns()``.
+
+ For these constructs the names reported by ``cursor.description`` are
+ generated by the database and don't correspond to the names given to
+ ``.columns()``, so the handlers have to be matched positionally.
+
+ See #13479
+
+ """
+
+ __only_on__ = "oracle"
+ __backend__ = True
+
+ @classmethod
+ def define_tables(cls, metadata):
+ Table(
+ "data_table",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("data", sqltypes.JSON, nullable=True),
+ )
+
+ @testing.requires.json_type
+ def test_json_bound_literal(self, connection):
+ """the JSON value arrives as a plain string bound parameter, with
+ no JSON column involved at all"""
+
+ value = {"json": {"foo": "bar"}, "recs": ["one", "two"]}
+
+ stmt = text("select :p from dual").columns(column("jj", sqltypes.JSON))
+ eq_(connection.scalar(stmt, {"p": json.dumps(value)}), value)
+
+ @testing.requires.json_type
+ def test_json_column(self, connection):
+ value = {"json": {"foo": "bar"}}
+ connection.execute(
+ self.tables.data_table.insert(), {"id": 1, "data": value}
+ )
+
+ stmt = text("select data from data_table").columns(
+ column("jj", sqltypes.JSON)
+ )
+ eq_(connection.scalar(stmt), value)
+
+ @testing.requires.json_type
+ def test_json_null_value(self, connection):
+ connection.execute(
+ self.tables.data_table.insert(), {"id": 1, "data": None}
+ )
+
+ stmt = text("select data from data_table").columns(
+ column("jj", sqltypes.JSON)
+ )
+ eq_(connection.scalar(stmt), None)
+
+ @testing.requires.json_type
+ def test_json_clob_expression(self, connection):
+ """a JSON expression that the database reports as a LOB"""
+
+ value = {"json": {"foo": "bar"}, "recs": ["one", "two"]}
+
+ stmt = text("select to_clob(:p) from dual").columns(
+ column("jj", sqltypes.JSON)
+ )
+ eq_(connection.scalar(stmt, {"p": json.dumps(value)}), value)
+
+ def test_json_blob_bound_literal(self, connection):
+ value = {"json": {"foo": "bar"}}
+
+ stmt = text("select :p from dual").columns(
+ column("jj", oracle.JSON(use_blob=True))
+ )
+ eq_(connection.scalar(stmt, {"p": json.dumps(value)}), value)
+
+ @testing.requires.json_type
+ def test_positional_alignment_mixed_types(self, connection):
+ """handlers have to line up with the cursor's columns positionally,
+ including for columns that have no handler of their own"""
+
+ value = {"json": {"foo": "bar"}}
+
+ stmt = text("select 'x', :p, 1, 2.5, 'y' from dual").columns(
+ column("a", String),
+ column("jj", sqltypes.JSON),
+ column("i", Integer),
+ column("n", Numeric(asdecimal=True)),
+ column("b", String),
+ )
+ eq_(
+ connection.execute(stmt, {"p": json.dumps(value)}).all(),
+ [("x", value, 1, decimal.Decimal("2.5"), "y")],
+ )
+
+ def test_numeric_positional(self, connection):
+ """Numeric is normally handled by the connection-level handler;
+ assert the cursor-level handler doesn't misalign it"""
+
+ stmt = text("select 1.5, 2.5 from dual").columns(
+ column("a", Numeric(asdecimal=False)),
+ column("b", Numeric(asdecimal=True)),
+ )
+ eq_(connection.execute(stmt).all(), [(1.5, decimal.Decimal("2.5"))])
+
+ @testing.requires.json_type
+ def test_compiled_select_still_matches_by_name(self, connection):
+ """the non-textual case continues to match handlers by the
+ rendered column name"""
+
+ value = {"json": {"foo": "bar"}}
+ eq_(
+ connection.scalar(select(literal(value, sqltypes.JSON))),
+ value,
+ )
+
+
+class JSONDeserializerTest(fixtures.TablesTest):
+ """test that a custom ``json_deserializer`` is honored by each of the
+ branches of the JSON outputtypehandler.
+
+ """
+
+ __only_on__ = "oracle"
+ __backend__ = True
+ __requires__ = ("json_type",)
+
+ @classmethod
+ def define_tables(cls, metadata):
+ Table(
+ "json_deser_data",
+ metadata,
+ Column("id", Integer, primary_key=True),
+ Column("data", sqltypes.JSON),
+ )
+
+ @testing.fixture
+ def json_connection(self, testing_engine):
+ def loads(value):
+ return ("deserialized", json.loads(value))
+
+ engine = testing_engine(options={"json_deserializer": loads})
+ with engine.connect() as conn:
+ yield conn
+
+ def test_native_json_column(self, json_connection):
+ """a native JSON column, which the driver would otherwise decode
+ on its own"""
+
+ value = {"json": {"foo": "bar"}}
+ json_connection.execute(
+ self.tables.json_deser_data.insert(), {"id": 1, "data": value}
+ )
+
+ eq_(
+ json_connection.scalar(select(self.tables.json_deser_data.c.data)),
+ ("deserialized", value),
+ )
+
+ def test_character_expression(self, json_connection):
+ value = {"json": {"foo": "bar"}}
+
+ stmt = text("select :p from dual").columns(column("jj", sqltypes.JSON))
+ eq_(
+ json_connection.scalar(stmt, {"p": json.dumps(value)}),
+ ("deserialized", value),
+ )
+
+ def test_lob_expression(self, json_connection):
+ value = {"json": {"foo": "bar"}}
+
+ stmt = text("select to_clob(:p) from dual").columns(
+ column("jj", sqltypes.JSON)
+ )
+ eq_(
+ json_connection.scalar(stmt, {"p": json.dumps(value)}),
+ ("deserialized", value),
+ )
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
