Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/dialects/oracle/cx

1# dialects/oracle/cx_oracle.py

3# <see AUTHORS file>

5# This module is part of SQLAlchemy and is released under

6# the MIT License: https://www.opensource.org/licenses/mit-license.php

7# mypy: ignore-errors

10r""".. dialect:: oracle+cx_oracle

11 :name: cx-Oracle

12 :dbapi: cx_oracle

13 :connectstring: oracle+cx_oracle://user:pass@hostname:port[/dbname][?service_name=<service>[&key=value&key=value...]]

14 :url: https://oracle.github.io/python-cx_Oracle/

16Description

17-----------

19cx_Oracle was the original driver for Oracle Database. It was superseded by

20python-oracledb which should be used instead.

22DSN vs. Hostname connections

23-----------------------------

25cx_Oracle provides several methods of indicating the target database. The

26dialect translates from a series of different URL forms.

28Hostname Connections with Easy Connect Syntax

29^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

31Given a hostname, port and service name of the target database, for example

32from Oracle Database's Easy Connect syntax then connect in SQLAlchemy using the

33``service_name`` query string parameter::

35 engine = create_engine(

36 "oracle+cx_oracle://scott:tiger@hostname:port?service_name=myservice&encoding=UTF-8&nencoding=UTF-8"

37 )

39Note that the default driver value for encoding and nencoding was changed to

40“UTF-8” in cx_Oracle 8.0 so these parameters can be omitted when using that

41version, or later.

43To use a full Easy Connect string, pass it as the ``dsn`` key value in a

44:paramref:`_sa.create_engine.connect_args` dictionary::

46 import cx_Oracle

48 e = create_engine(

49 "oracle+cx_oracle://@",

50 connect_args={

51 "user": "scott",

52 "password": "tiger",

53 "dsn": "hostname:port/myservice?transport_connect_timeout=30&expire_time=60",

54 },

55 )

57Connections with tnsnames.ora or to Oracle Autonomous Database

58^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

60Alternatively, if no port, database name, or service name is provided, the

61dialect will use an Oracle Database DSN "connection string". This takes the

62"hostname" portion of the URL as the data source name. For example, if the

63``tnsnames.ora`` file contains a TNS Alias of ``myalias`` as below:

65.. sourcecode:: text

67 myalias =

68 (DESCRIPTION =

69 (ADDRESS = (PROTOCOL = TCP)(HOST = mymachine.example.com)(PORT = 1521))

70 (CONNECT_DATA =

71 (SERVER = DEDICATED)

72 (SERVICE_NAME = orclpdb1)

73 )

74 )

76The cx_Oracle dialect connects to this database service when ``myalias`` is the

77hostname portion of the URL, without specifying a port, database name or

78``service_name``::

80 engine = create_engine("oracle+cx_oracle://scott:tiger@myalias")

82Users of Oracle Autonomous Database should use this syntax. If the database is

83configured for mutural TLS ("mTLS"), then you must also configure the cloud

84wallet as shown in cx_Oracle documentation `Connecting to Autononmous Databases

85<https://cx-oracle.readthedocs.io/en/latest/user_guide/connection_handling.html#autonomousdb>`_.

87SID Connections

88^^^^^^^^^^^^^^^

90To use Oracle Database's obsolete System Identifier connection syntax, the SID

91can be passed in a "database name" portion of the URL::

93 engine = create_engine(

94 "oracle+cx_oracle://scott:tiger@hostname:port/dbname"

95 )

97Above, the DSN passed to cx_Oracle is created by ``cx_Oracle.makedsn()`` as

98follows::

100 >>> import cx_Oracle

101 >>> cx_Oracle.makedsn("hostname", 1521, sid="dbname")

102 '(DESCRIPTION=(ADDRESS=(PROTOCOL=TCP)(HOST=hostname)(PORT=1521))(CONNECT_DATA=(SID=dbname)))'

103

104Note that although the SQLAlchemy syntax ``hostname:port/dbname`` looks like

105Oracle's Easy Connect syntax it is different. It uses a SID in place of the

106service name required by Easy Connect. The Easy Connect syntax does not

107support SIDs.

108

109Passing cx_Oracle connect arguments

110-----------------------------------

111

112Additional connection arguments can usually be passed via the URL query string;

113particular symbols like ``SYSDBA`` are intercepted and converted to the correct

114symbol::

115

116 e = create_engine(

117 "oracle+cx_oracle://user:pass@dsn?encoding=UTF-8&nencoding=UTF-8&mode=SYSDBA&events=true"

118 )

119

120To pass arguments directly to ``.connect()`` without using the query

121string, use the :paramref:`_sa.create_engine.connect_args` dictionary.

122Any cx_Oracle parameter value and/or constant may be passed, such as::

123

124 import cx_Oracle

125

126 e = create_engine(

127 "oracle+cx_oracle://user:pass@dsn",

128 connect_args={

129 "encoding": "UTF-8",

130 "nencoding": "UTF-8",

131 "mode": cx_Oracle.SYSDBA,

132 "events": True,

133 },

134 )

135

136Note that the default driver value for ``encoding`` and ``nencoding`` was

137changed to "UTF-8" in cx_Oracle 8.0 so these parameters can be omitted when

138using that version, or later.

139

140Options consumed by the SQLAlchemy cx_Oracle dialect outside of the driver

141--------------------------------------------------------------------------

142

143There are also options that are consumed by the SQLAlchemy cx_oracle dialect

144itself. These options are always passed directly to :func:`_sa.create_engine`

145, such as::

146

147 e = create_engine(

148 "oracle+cx_oracle://user:pass@dsn", coerce_to_decimal=False

149 )

150

151The parameters accepted by the cx_oracle dialect are as follows:

152

153* ``arraysize`` - set the cx_oracle.arraysize value on cursors; defaults

154 to ``None``, indicating that the driver default should be used (typically

155 the value is 100). This setting controls how many rows are buffered when

156 fetching rows, and can have a significant effect on performance when

157 modified.

158

159 .. versionchanged:: 2.0.26 - changed the default value from 50 to None,

160 to use the default value of the driver itself.

161

162* ``auto_convert_lobs`` - defaults to True; See :ref:`cx_oracle_lob`.

163

164* ``coerce_to_decimal`` - see :ref:`cx_oracle_numeric` for detail.

165

166* ``encoding_errors`` - see :ref:`cx_oracle_unicode_encoding_errors` for detail.

167

168.. _cx_oracle_sessionpool:

169

170Using cx_Oracle SessionPool

171---------------------------

172

173The cx_Oracle driver provides its own connection pool implementation that may

174be used in place of SQLAlchemy's pooling functionality. The driver pool

175supports Oracle Database features such dead connection detection, connection

176draining for planned database downtime, support for Oracle Application

177Continuity and Transparent Application Continuity, and gives support for

178Database Resident Connection Pooling (DRCP).

179

180Using the driver pool can be achieved by using the

181:paramref:`_sa.create_engine.creator` parameter to provide a function that

182returns a new connection, along with setting

183:paramref:`_sa.create_engine.pool_class` to ``NullPool`` to disable

184SQLAlchemy's pooling::

185

186 import cx_Oracle

187 from sqlalchemy import create_engine

188 from sqlalchemy.pool import NullPool

189

190 pool = cx_Oracle.SessionPool(

191 user="scott",

192 password="tiger",

193 dsn="orclpdb",

194 min=1,

195 max=4,

196 increment=1,

197 threaded=True,

198 encoding="UTF-8",

199 nencoding="UTF-8",

200 )

201

202 engine = create_engine(

203 "oracle+cx_oracle://", creator=pool.acquire, poolclass=NullPool

204 )

205

206The above engine may then be used normally where cx_Oracle's pool handles

207connection pooling::

208

209 with engine.connect() as conn:

210 print(conn.scalar("select 1 from dual"))

211

212As well as providing a scalable solution for multi-user applications, the

213cx_Oracle session pool supports some Oracle features such as DRCP and

214`Application Continuity

215<https://cx-oracle.readthedocs.io/en/latest/user_guide/ha.html#application-continuity-ac>`_.

216

217Note that the pool creation parameters ``threaded``, ``encoding`` and

218``nencoding`` were deprecated in later cx_Oracle releases.

219

220Using Oracle Database Resident Connection Pooling (DRCP)

221--------------------------------------------------------

222

223When using Oracle Database's DRCP, the best practice is to pass a connection

224class and "purity" when acquiring a connection from the SessionPool. Refer to

225the `cx_Oracle DRCP documentation

226<https://cx-oracle.readthedocs.io/en/latest/user_guide/connection_handling.html#database-resident-connection-pooling-drcp>`_.

227

228This can be achieved by wrapping ``pool.acquire()``::

229

230 import cx_Oracle

231 from sqlalchemy import create_engine

232 from sqlalchemy.pool import NullPool

233

234 pool = cx_Oracle.SessionPool(

235 user="scott",

236 password="tiger",

237 dsn="orclpdb",

238 min=2,

239 max=5,

240 increment=1,

241 threaded=True,

242 encoding="UTF-8",

243 nencoding="UTF-8",

244 )

245

246

247 def creator():

248 return pool.acquire(

249 cclass="MYCLASS", purity=cx_Oracle.ATTR_PURITY_SELF

250 )

251

252

253 engine = create_engine(

254 "oracle+cx_oracle://", creator=creator, poolclass=NullPool

255 )

256

257The above engine may then be used normally where cx_Oracle handles session

258pooling and Oracle Database additionally uses DRCP::

259

260 with engine.connect() as conn:

261 print(conn.scalar("select 1 from dual"))

262

263.. _cx_oracle_unicode:

264

265Unicode

266-------

267

268As is the case for all DBAPIs under Python 3, all strings are inherently

269Unicode strings. In all cases however, the driver requires an explicit

270encoding configuration.

271

272Ensuring the Correct Client Encoding

273^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

274

275The long accepted standard for establishing client encoding for nearly all

276Oracle Database related software is via the `NLS_LANG

277<https://www.oracle.com/database/technologies/faq-nls-lang.html>`_ environment

278variable. Older versions of cx_Oracle use this environment variable as the

279source of its encoding configuration. The format of this variable is

280Territory_Country.CharacterSet; a typical value would be

281``AMERICAN_AMERICA.AL32UTF8``. cx_Oracle version 8 and later use the character

282set "UTF-8" by default, and ignore the character set component of NLS_LANG.

283

284The cx_Oracle driver also supported a programmatic alternative which is to pass

285the ``encoding`` and ``nencoding`` parameters directly to its ``.connect()``

286function. These can be present in the URL as follows::

287

288 engine = create_engine(

289 "oracle+cx_oracle://scott:tiger@tnsalias?encoding=UTF-8&nencoding=UTF-8"

290 )

291

292For the meaning of the ``encoding`` and ``nencoding`` parameters, please

293consult

294`Characters Sets and National Language Support (NLS) <https://cx-oracle.readthedocs.io/en/latest/user_guide/globalization.html#globalization>`_.

295

296.. seealso::

297

298 `Characters Sets and National Language Support (NLS) <https://cx-oracle.readthedocs.io/en/latest/user_guide/globalization.html#globalization>`_

299 - in the cx_Oracle documentation.

300

301

302Unicode-specific Column datatypes

303^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

304

305The Core expression language handles unicode data by use of the

306:class:`.Unicode` and :class:`.UnicodeText` datatypes. These types correspond

307to the VARCHAR2 and CLOB Oracle Database datatypes by default. When using

308these datatypes with Unicode data, it is expected that the database is

309configured with a Unicode-aware character set, as well as that the ``NLS_LANG``

310environment variable is set appropriately (this applies to older versions of

311cx_Oracle), so that the VARCHAR2 and CLOB datatypes can accommodate the data.

312

313In the case that Oracle Database is not configured with a Unicode character

314set, the two options are to use the :class:`_types.NCHAR` and

315:class:`_oracle.NCLOB` datatypes explicitly, or to pass the flag

316``use_nchar_for_unicode=True`` to :func:`_sa.create_engine`, which will cause

317the SQLAlchemy dialect to use NCHAR/NCLOB for the :class:`.Unicode` /

318:class:`.UnicodeText` datatypes instead of VARCHAR/CLOB.

319

320.. _cx_oracle_unicode_encoding_errors:

321

322Encoding Errors

323^^^^^^^^^^^^^^^

324

325For the unusual case that data in Oracle Database is present with a broken

326encoding, the dialect accepts a parameter ``encoding_errors`` which will be

327passed to Unicode decoding functions in order to affect how decoding errors are

328handled. The value is ultimately consumed by the Python `decode

329<https://docs.python.org/3/library/stdtypes.html#bytes.decode>`_ function, and

330is passed both via cx_Oracle's ``encodingErrors`` parameter consumed by

331``Cursor.var()``, as well as SQLAlchemy's own decoding function, as the

332cx_Oracle dialect makes use of both under different circumstances.

333

334.. _cx_oracle_setinputsizes:

335

336Fine grained control over cx_Oracle data binding performance with setinputsizes

337-------------------------------------------------------------------------------

338

339The cx_Oracle DBAPI has a deep and fundamental reliance upon the usage of the

340DBAPI ``setinputsizes()`` call. The purpose of this call is to establish the

341datatypes that are bound to a SQL statement for Python values being passed as

342parameters. While virtually no other DBAPI assigns any use to the

343``setinputsizes()`` call, the cx_Oracle DBAPI relies upon it heavily in its

344interactions with the Oracle Database client interface, and in some scenarios

345it is not possible for SQLAlchemy to know exactly how data should be bound, as

346some settings can cause profoundly different performance characteristics, while

347altering the type coercion behavior at the same time.

348

349Users of the cx_Oracle dialect are **strongly encouraged** to read through

350cx_Oracle's list of built-in datatype symbols at

351https://cx-oracle.readthedocs.io/en/latest/api_manual/module.html#database-types.

352Note that in some cases, significant performance degradation can occur when

353using these types vs. not, in particular when specifying ``cx_Oracle.CLOB``.

354

355On the SQLAlchemy side, the :meth:`.DialectEvents.do_setinputsizes` event can

356be used both for runtime visibility (e.g. logging) of the setinputsizes step as

357well as to fully control how ``setinputsizes()`` is used on a per-statement

358basis.

359

360Example 1 - logging all setinputsizes calls

361^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

362

363The following example illustrates how to log the intermediary values from a

364SQLAlchemy perspective before they are converted to the raw ``setinputsizes()``

365parameter dictionary. The keys of the dictionary are :class:`.BindParameter`

366objects which have a ``.key`` and a ``.type`` attribute::

367

368 from sqlalchemy import create_engine, event

369

370 engine = create_engine("oracle+cx_oracle://scott:tiger@host/xe")

371

372

373 @event.listens_for(engine, "do_setinputsizes")

374 def _log_setinputsizes(inputsizes, cursor, statement, parameters, context):

375 for bindparam, dbapitype in inputsizes.items():

376 log.info(

377 "Bound parameter name: %s SQLAlchemy type: %r DBAPI object: %s",

378 bindparam.key,

379 bindparam.type,

380 dbapitype,

381 )

382

383Example 2 - remove all bindings to CLOB

384^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

385

386The ``CLOB`` datatype in cx_Oracle incurs a significant performance overhead,

387however is set by default for the ``Text`` type within the SQLAlchemy 1.2

388series. This setting can be modified as follows::

389

390 from sqlalchemy import create_engine, event

391 from cx_Oracle import CLOB

392

393 engine = create_engine("oracle+cx_oracle://scott:tiger@host/xe")

394

395

396 @event.listens_for(engine, "do_setinputsizes")

397 def _remove_clob(inputsizes, cursor, statement, parameters, context):

398 for bindparam, dbapitype in list(inputsizes.items()):

399 if dbapitype is CLOB:

400 del inputsizes[bindparam]

401

402.. _cx_oracle_lob:

403

404LOB Datatypes

405--------------

406

407LOB datatypes refer to the "large object" datatypes such as CLOB, NCLOB and

408BLOB. Modern versions of cx_Oracle is optimized for these datatypes to be

409delivered as a single buffer. As such, SQLAlchemy makes use of these newer type

410handlers by default.

411

412To disable the use of newer type handlers and deliver LOB objects as classic

413buffered objects with a ``read()`` method, the parameter

414``auto_convert_lobs=False`` may be passed to :func:`_sa.create_engine`,

415which takes place only engine-wide.

416

417.. _cx_oracle_returning:

418

419RETURNING Support

420-----------------

421

422The cx_Oracle dialect implements RETURNING using OUT parameters.

423The dialect supports RETURNING fully.

424

425Two Phase Transactions Not Supported

426------------------------------------

427

428Two phase transactions are **not supported** under cx_Oracle due to poor driver

429support. The newer :ref:`oracledb` dialect however **does** support two phase

430transactions.

431

432.. _cx_oracle_numeric:

433

434Precision Numerics

435------------------

436

437SQLAlchemy's numeric types can handle receiving and returning values as Python

438``Decimal`` objects or float objects. When a :class:`.Numeric` object, or a

439subclass such as :class:`.Float`, :class:`_oracle.DOUBLE_PRECISION` etc. is in

440use, the :paramref:`.Numeric.asdecimal` flag determines if values should be

441coerced to ``Decimal`` upon return, or returned as float objects. To make

442matters more complicated under Oracle Database, the ``NUMBER`` type can also

443represent integer values if the "scale" is zero, so the Oracle

444Database-specific :class:`_oracle.NUMBER` type takes this into account as well.

445

446The cx_Oracle dialect makes extensive use of connection- and cursor-level

447"outputtypehandler" callables in order to coerce numeric values as requested.

448These callables are specific to the specific flavor of :class:`.Numeric` in

449use, as well as if no SQLAlchemy typing objects are present. There are

450observed scenarios where Oracle Database may send incomplete or ambiguous

451information about the numeric types being returned, such as a query where the

452numeric types are buried under multiple levels of subquery. The type handlers

453do their best to make the right decision in all cases, deferring to the

454underlying cx_Oracle DBAPI for all those cases where the driver can make the

455best decision.

456

457When no typing objects are present, as when executing plain SQL strings, a

458default "outputtypehandler" is present which will generally return numeric

459values which specify precision and scale as Python ``Decimal`` objects. To

460disable this coercion to decimal for performance reasons, pass the flag

461``coerce_to_decimal=False`` to :func:`_sa.create_engine`::

462

463 engine = create_engine("oracle+cx_oracle://dsn", coerce_to_decimal=False)

464

465The ``coerce_to_decimal`` flag only impacts the results of plain string

466SQL statements that are not otherwise associated with a :class:`.Numeric`

467SQLAlchemy type (or a subclass of such).

468

469""" # noqa

470

471from __future__ import annotations

472

473import collections

474import decimal

475import json

476import random

477import re

478

479from . import base as oracle

480from .base import OracleCompiler

481from .base import OracleDialect

482from .base import OracleExecutionContext

483from .json import JSON

484from .types import _OracleDateLiteralRender

485from ... import exc

486from ... import util

487from ...engine import cursor as _cursor

488from ...engine import interfaces

489from ...engine import processors

490from ...sql import sqltypes

491from ...sql._typing import is_sql_compiler

492from ...sql.base import NO_ARG

493from ...sql.sqltypes import Boolean

494

495# source:

496# https://github.com/oracle/python-cx_Oracle/issues/596#issuecomment-999243649

497_CX_ORACLE_MAGIC_LOB_SIZE = 131072

498

499# largest JSON we can deserialize if we are not using

500# DB_TYPE_JSON

501_CX_ORACLE_MAX_JSON_CONVERTED = 32767

502

503

504class _OracleJson(JSON):

505 def get_dbapi_type(self, dbapi):

506 return dbapi.DB_TYPE_JSON

507

508 def _should_use_blob(self, dialect):

509 use_blob = (

510 not dialect._supports_oracle_json

511 if self.use_blob is NO_ARG

512 else self.use_blob

513 )

514

515 return use_blob

516

517 def bind_processor(self, dialect):

518

519 if self._should_use_blob(dialect):

520

521 DBAPIBinary = dialect.dbapi.Binary

522

523 def string_process(value):

524 if value is not None:

525 # utf-8 is standard for oracledb

526 # https://python-oracledb.readthedocs.io/en/latest/user_guide/globalization.html#setting-the-client-character-set # noqa: E501

527 return DBAPIBinary(value.encode("utf-8"))

528 else:

529 return None

530

531 else:

532 string_process = None

533

534 json_serializer = dialect._json_serializer or json.dumps

535

536 return self._make_bind_processor(string_process, json_serializer)

537

538 def result_processor(self, dialect, coltype):

539 if self._should_use_blob(dialect):

540 # for plain BLOB, use traditional binary decode + json.loads()

541 string_process = self._str_impl.result_processor(dialect, coltype)

542 json_deserializer = dialect._json_deserializer or json.loads

543

544 def process(value):

545 if value is None:

546 return None

547 if string_process:

548 value = string_process(value)

549 return json_deserializer(value)

550

551 return process

552

553 else:

554 # not BLOB; the value is decoded by an outputtypehandler

555 # rather than here. see _cx_oracle_outputtypehandler() below

556 return None

557

558 def _cx_oracle_outputtypehandler(self, dialect):

559 """Establish an outputtypehandler for JSON result columns.

560

561 The dialect makes use of two distinct outputtypehandlers. One is

562 connection-wide, set up by the dialect's

563 _generate_connection_outputtype_handler(); it sees every column of

564 every statement and knows nothing of SQLAlchemy-side types, only of

565 the DBAPI type the database reports. The other is the one returned

566 here, which the execution context's

567 _generate_cursor_outputtype_handler() installs on the cursor for one

568 particular statement, and which applies only to those result columns

569 that SQLAlchemy knows to be of JSON type.

570

571 The two do not compose. The driver consults the cursor's handler if

572 one is present, and the connection's otherwise, never both. The

573 cursor-level handler assembled by the execution context therefore

574 delegates to the connection-level handler explicitly for the columns

575 it has no handler of its own for, and the DB_TYPE_JSON case below

576 has to repeat what the connection-level handler does rather than

577 defer to it.

578

579 """

580

581 if self._should_use_blob(dialect):

582 # BLOB is decoded by result_processor() instead

583 return None

584

585 cx_Oracle = dialect.dbapi

586 json_deserializer = dialect._json_deserializer

587

588 def handler(cursor, name, default_type, size, precision, scale):

589 if default_type is cx_Oracle.DB_TYPE_JSON:

590 # a native JSON column. the driver decodes these into

591 # Python objects on its own, so a handler is only needed in

592 # order to route the value through a user-supplied

593 # deserializer, receiving it as text to do so. this repeats

594 # the connection-level handler, as noted above

595 if json_deserializer is not None:

596 return cursor.var(

597 cx_Oracle.DB_TYPE_VARCHAR,

598 _CX_ORACLE_MAX_JSON_CONVERTED,

599 cursor.arraysize,

600 outconverter=json_deserializer,

601 )

602 else:

603 return None

604 elif default_type in (

605 cx_Oracle.DB_TYPE_VARCHAR,

606 cx_Oracle.DB_TYPE_NVARCHAR,

607 cx_Oracle.DB_TYPE_LONG,

608 cx_Oracle.DB_TYPE_LONG_NVARCHAR,

609 ):

610 # a JSON-typed expression that isn't a native JSON column,

611 # such as a bound parameter or a string expression selected

612 # directly. the database reports it as ordinary character

613 # data and won't decode it, so decode it here. the

614 # connection-level handler can't do this as it has no way to

615 # know the expression was intended as JSON

616 return cursor.var(

617 default_type,

618 size,

619 cursor.arraysize,

620 outconverter=json_deserializer or json.loads,

621 **dialect._cursor_var_unicode_kwargs,

622 )

623 elif default_type in (cx_Oracle.CLOB, cx_Oracle.NCLOB):

624 # same as the character case above, except the database

625 # reports the expression as a LOB; read it as a string

626 # first, then decode

627 return dialect._cursor_lob_as_str_var(

628 cursor, default_type, json_deserializer or json.loads

629 )

630 else:

631 return None

632

633 return handler

634

635

636class _OracleInteger(sqltypes.Integer):

637 def get_dbapi_type(self, dbapi):

638 # see https://github.com/oracle/python-cx_Oracle/issues/

639 # 208#issuecomment-409715955

640 return int

641

642 def _cx_oracle_var(self, dialect, cursor, arraysize=None):

643 cx_Oracle = dialect.dbapi

644 return cursor.var(

645 cx_Oracle.STRING,

646 255,

647 arraysize=arraysize if arraysize is not None else cursor.arraysize,

648 outconverter=int,

649 )

650

651 def _cx_oracle_outputtypehandler(self, dialect):

652 def handler(cursor, name, default_type, size, precision, scale):

653 return self._cx_oracle_var(dialect, cursor)

654

655 return handler

656

657

658class _OracleNumericCommon(sqltypes.NumericCommon, sqltypes.TypeEngine):

659 is_number = False

660

661 def bind_processor(self, dialect):

662 if self.scale == 0:

663 return None

664 elif self.asdecimal:

665 processor = processors.to_decimal_processor_factory(

666 decimal.Decimal, self._effective_decimal_return_scale

667 )

668

669 def process(value):

670 if isinstance(value, (int, float)):

671 return processor(value)

672 elif value is not None and value.is_infinite():

673 return float(value)

674 else:

675 return value

676

677 return process

678 else:

679 return processors.to_float

680

681 def result_processor(self, dialect, coltype):

682 return None

683

684 def _cx_oracle_outputtypehandler(self, dialect):

685 cx_Oracle = dialect.dbapi

686

687 def handler(cursor, name, default_type, size, precision, scale):

688 outconverter = None

689

690 if precision:

691 if self.asdecimal:

692 if default_type == cx_Oracle.NATIVE_FLOAT:

693 # receiving float and doing Decimal after the fact

694 # allows for float("inf") to be handled

695 type_ = default_type

696 outconverter = decimal.Decimal

697 else:

698 type_ = decimal.Decimal

699 else:

700 if self.is_number and scale == 0:

701 # integer. cx_Oracle is observed to handle the widest

702 # variety of ints when no directives are passed,

703 # from 5.2 to 7.0. See [ticket:4457]

704 return None

705 else:

706 type_ = cx_Oracle.NATIVE_FLOAT

707

708 else:

709 if self.asdecimal:

710 if default_type == cx_Oracle.NATIVE_FLOAT:

711 type_ = default_type

712 outconverter = decimal.Decimal

713 else:

714 type_ = decimal.Decimal

715 else:

716 if self.is_number and scale == 0:

717 # integer. cx_Oracle is observed to handle the widest

718 # variety of ints when no directives are passed,

719 # from 5.2 to 7.0. See [ticket:4457]

720 return None

721 else:

722 type_ = cx_Oracle.NATIVE_FLOAT

723

724 return cursor.var(

725 type_,

726 255,

727 arraysize=cursor.arraysize,

728 outconverter=outconverter,

729 )

730

731 return handler

732

733

734class _OracleNumeric(_OracleNumericCommon, sqltypes.Numeric):

735 pass

736

737

738class _OracleFloat(_OracleNumericCommon, sqltypes.Float):

739 pass

740

741

742class _OracleUUID(sqltypes.Uuid):

743 def get_dbapi_type(self, dbapi):

744 return dbapi.STRING

745

746

747class _OracleBinaryFloat(_OracleNumericCommon):

748 def get_dbapi_type(self, dbapi):

749 return dbapi.NATIVE_FLOAT

750

751

752class _OracleBINARY_FLOAT(_OracleBinaryFloat, oracle.BINARY_FLOAT):

753 pass

754

755

756class _OracleBINARY_DOUBLE(_OracleBinaryFloat, oracle.BINARY_DOUBLE):

757 pass

758

759

760class _OracleNUMBER(_OracleNumericCommon, sqltypes.Numeric):

761 is_number = True

762

763

764class _CXOracleDate(oracle._OracleDate):

765 def bind_processor(self, dialect):

766 return None

767

768 def result_processor(self, dialect, coltype):

769 def process(value):

770 if value is not None:

771 return value.date()

772 else:

773 return value

774

775 return process

776

777

778class _CXOracleTIMESTAMP(_OracleDateLiteralRender, sqltypes.TIMESTAMP):

779 def literal_processor(self, dialect):

780 return self._literal_processor_datetime(dialect)

781

782

783class _LOBDataType:

784 pass

785

786

787# TODO: the names used across CHAR / VARCHAR / NCHAR / NVARCHAR

788# here are inconsistent and not very good

789class _OracleChar(sqltypes.CHAR):

790 def get_dbapi_type(self, dbapi):

791 return dbapi.FIXED_CHAR

792

793

794class _OracleNChar(sqltypes.NCHAR):

795 def get_dbapi_type(self, dbapi):

796 return dbapi.FIXED_NCHAR

797

798

799class _OracleUnicodeStringNCHAR(oracle.NVARCHAR2):

800 def get_dbapi_type(self, dbapi):

801 return dbapi.NCHAR

802

803

804class _OracleUnicodeStringCHAR(sqltypes.Unicode):

805 def get_dbapi_type(self, dbapi):

806 return dbapi.LONG_STRING

807

808

809class _OracleUnicodeTextNCLOB(_LOBDataType, oracle.NCLOB):

810 def get_dbapi_type(self, dbapi):

811 # previously, this was dbapi.NCLOB.

812 # DB_TYPE_NVARCHAR will instead be passed to setinputsizes()

813 # when this datatype is used.

814 return dbapi.DB_TYPE_NVARCHAR

815

816

817class _OracleUnicodeTextCLOB(_LOBDataType, sqltypes.UnicodeText):

818 def get_dbapi_type(self, dbapi):

819 # previously, this was dbapi.CLOB.

820 # DB_TYPE_NVARCHAR will instead be passed to setinputsizes()

821 # when this datatype is used.

822 return dbapi.DB_TYPE_NVARCHAR

823

824

825class _OracleText(_LOBDataType, sqltypes.Text):

826 def get_dbapi_type(self, dbapi):

827 # previously, this was dbapi.CLOB.

828 # DB_TYPE_NVARCHAR will instead be passed to setinputsizes()

829 # when this datatype is used.

830 return dbapi.DB_TYPE_NVARCHAR

831

832

833class _OracleLong(_LOBDataType, oracle.LONG):

834 def get_dbapi_type(self, dbapi):

835 return dbapi.LONG_STRING

836

837

838class _OracleString(sqltypes.String):

839 pass

840

841

842class _OracleEnum(sqltypes.Enum):

843 def bind_processor(self, dialect):

844 enum_proc = sqltypes.Enum.bind_processor(self, dialect)

845

846 def process(value):

847 raw_str = enum_proc(value)

848 return raw_str

849

850 return process

851

852

853class _OracleBinary(_LOBDataType, sqltypes.LargeBinary):

854 def get_dbapi_type(self, dbapi):

855 # previously, this was dbapi.BLOB.

856 # DB_TYPE_RAW will instead be passed to setinputsizes()

857 # when this datatype is used.

858 return dbapi.DB_TYPE_RAW

859

860 def bind_processor(self, dialect):

861 return None

862

863 def result_processor(self, dialect, coltype):

864 if not dialect.auto_convert_lobs:

865 return None

866 else:

867 return super().result_processor(dialect, coltype)

868

869

870class _OracleInterval(oracle.INTERVAL):

871 def get_dbapi_type(self, dbapi):

872 return dbapi.INTERVAL

873

874

875class _OracleRaw(oracle.RAW):

876 pass

877

878

879class _OracleRowid(oracle.ROWID):

880 def get_dbapi_type(self, dbapi):

881 return dbapi.ROWID

882

883

884class OracleCompiler_cx_oracle(OracleCompiler):

885 _oracle_cx_sql_compiler = True

886

887 _oracle_returning = False

888

889 # Oracle bind names can't start with digits or underscores.

890 # currently we rely upon Oracle-specific quoting of bind names in most

891 # cases. however for expanding params, the escape chars are used.

892 # see #8708

893 bindname_escape_characters = util.immutabledict(

894 {

895 "%": "P",

896 "(": "A",

897 ")": "Z",

898 ":": "C",

899 ".": "C",

900 "[": "C",

901 "]": "C",

902 " ": "C",

903 "\\": "C",

904 "/": "C",

905 "?": "C",

906 }

907 )

908

909 def bindparam_string(self, name, **kw):

910 quote = getattr(name, "quote", None)

911 if (

912 quote is True

913 or quote is not False

914 and self.preparer._bindparam_requires_quotes(name)

915 # bind param quoting for Oracle doesn't work with post_compile

916 # params. For those, the default bindparam_string will escape

917 # special chars, and the appending of a number "_1" etc. will

918 # take care of reserved words

919 and not kw.get("post_compile", False)

920 ):

921 # interesting to note about expanding parameters - since the

922 # new parameters take the form <paramname>_<int>, at least if

923 # they are originally formed from reserved words, they no longer

924 # need quoting :). names that include illegal characters

925 # won't work however.

926 quoted_name = '"%s"' % name

927 kw["escaped_from"] = name

928 name = quoted_name

929 return OracleCompiler.bindparam_string(self, name, **kw)

930

931 # TODO: we could likely do away with quoting altogether for

932 # Oracle parameters and use the custom escaping here

933 escaped_from = kw.get("escaped_from", None)

934 if not escaped_from:

935 if self._bind_translate_re.search(name):

936 # not quite the translate use case as we want to

937 # also get a quick boolean if we even found

938 # unusual characters in the name

939 new_name = self._bind_translate_re.sub(

940 lambda m: self._bind_translate_chars[m.group(0)],

941 name,

942 )

943 if new_name[0].isdigit() or new_name[0] == "_":

944 new_name = "D" + new_name

945 kw["escaped_from"] = name

946 name = new_name

947 elif name[0].isdigit() or name[0] == "_":

948 new_name = "D" + name

949 kw["escaped_from"] = name

950 name = new_name

951

952 return OracleCompiler.bindparam_string(self, name, **kw)

953

954

955class OracleExecutionContext_cx_oracle(OracleExecutionContext):

956 out_parameters = None

957

958 def _generate_out_parameter_vars(self):

959 # check for has_out_parameters or RETURNING, create cx_Oracle.var

960 # objects if so

961 if self.compiled.has_out_parameters or self.compiled._oracle_returning:

962 out_parameters = self.out_parameters

963 assert out_parameters is not None

964

965 len_params = len(self.parameters)

966

967 quoted_bind_names = self.compiled.escaped_bind_names

968 for bindparam in self.compiled.binds.values():

969 if bindparam.isoutparam:

970 name = self.compiled.bind_names[bindparam]

971 type_impl = bindparam.type.dialect_impl(self.dialect)

972

973 if hasattr(type_impl, "_cx_oracle_var"):

974 out_parameters[name] = type_impl._cx_oracle_var(

975 self.dialect, self.cursor, arraysize=len_params

976 )

977 else:

978 dbtype = type_impl.get_dbapi_type(self.dialect.dbapi)

979

980 cx_Oracle = self.dialect.dbapi

981

982 assert cx_Oracle is not None

983

984 if dbtype is None:

985 raise exc.InvalidRequestError(

986 "Cannot create out parameter for "

987 "parameter "

988 "%r - its type %r is not supported by"

989 " cx_oracle" % (bindparam.key, bindparam.type)

990 )

991

992 # note this is an OUT parameter. Using

993 # non-LOB datavalues with large unicode-holding

994 # values causes the failure (both cx_Oracle and

995 # oracledb):

996 # ORA-22835: Buffer too small for CLOB to CHAR or

997 # BLOB to RAW conversion (actual: 16507,

998 # maximum: 4000)

999 # [SQL: INSERT INTO long_text (x, y, z) VALUES

1000 # (:x, :y, :z) RETURNING long_text.x, long_text.y,

1001 # long_text.z INTO :ret_0, :ret_1, :ret_2]

1002 # so even for DB_TYPE_NVARCHAR we convert to a LOB

1003

1004 if isinstance(type_impl, _LOBDataType):

1005 if dbtype == cx_Oracle.DB_TYPE_NVARCHAR:

1006 dbtype = cx_Oracle.NCLOB

1007 elif dbtype == cx_Oracle.DB_TYPE_RAW:

1008 dbtype = cx_Oracle.BLOB

1009 # other LOB types go in directly

1010

1011 out_parameters[name] = self.cursor.var(

1012 dbtype,

1013 # this is fine also in oracledb_async since

1014 # the driver will await the read coroutine

1015 outconverter=lambda value: value.read(),

1016 arraysize=len_params,

1017 )

1018 elif (

1019 isinstance(type_impl, _OracleNumericCommon)

1020 and type_impl.asdecimal

1021 ):

1022 out_parameters[name] = self.cursor.var(

1023 decimal.Decimal,

1024 arraysize=len_params,

1025 )

1026 elif isinstance(type_impl, Boolean):

1027 if self.dialect.supports_native_boolean:

1028 out_parameters[name] = self.cursor.var(

1029 cx_Oracle.BOOLEAN, arraysize=len_params

1030 )

1031 else:

1032 out_parameters[name] = self.cursor.var(

1033 cx_Oracle.NUMBER,

1034 arraysize=len_params,

1035 outconverter=bool,

1036 )

1037 else:

1038 out_parameters[name] = self.cursor.var(

1039 dbtype, arraysize=len_params

1040 )

1041

1042 for param in self.parameters:

1043 param[quoted_bind_names.get(name, name)] = (

1044 out_parameters[name]

1045 )

1046

1047 def _generate_cursor_outputtype_handler(self):

1048 assert isinstance(self.compiled, OracleCompiler)

1049

1050 # accumulate handlers positionally

1051 handlers = [

1052 type_._cached_custom_processor(

1053 self.dialect,

1054 "cx_oracle_outputtypehandler",

1055 self._get_cx_oracle_type_handler,

1056 )

1057 for _, _, _, type_ in self.compiled._result_columns

1058 ]

1059

1060 if not any(handlers):

1061 return

1062

1063 # a handler on the cursor replaces the connection-wide handler

1064 # outright rather than adding to it, so keep a reference to the

1065 # latter and delegate to it for columns we have no handler for

1066 default_handler = self._dbapi_connection.outputtypehandler

1067

1068 if self.compiled._textual_ordered_columns:

1069 # for a textual construct with positional columns, i.e.

1070 # text().columns() / tstring().columns(), match handlers

1071 # positionally instead of by name, since names are not

1072 # deterministic. See #13479

1073

1074 handler_queue = collections.deque(handlers)

1075

1076 def output_type_handler(

1077 cursor, name, default_type, size, precision, scale

1078 ):

1079 handler = handler_queue.popleft() if handler_queue else None

1080 if handler is not None:

1081 return handler(

1082 cursor, name, default_type, size, precision, scale

1083 )

1084 else:

1085 return default_handler(

1086 cursor, name, default_type, size, precision, scale

1087 )

1088

1089 else:

1090 output_handlers = {

1091 self.dialect.denormalize_name(rc[0]): handler

1092 for rc, handler in zip(self.compiled._result_columns, handlers)

1093 if handler

1094 }

1095

1096 def output_type_handler(

1097 cursor, name, default_type, size, precision, scale

1098 ):

1099 handler = output_handlers.get(name)

1100 if handler is not None:

1101 return handler(

1102 cursor, name, default_type, size, precision, scale

1103 )

1104 else:

1105 return default_handler(

1106 cursor, name, default_type, size, precision, scale

1107 )

1108

1109 self.cursor.outputtypehandler = output_type_handler

1110

1111 def _get_cx_oracle_type_handler(self, impl):

1112 if hasattr(impl, "_cx_oracle_outputtypehandler"):

1113 return impl._cx_oracle_outputtypehandler(self.dialect)

1114 else:

1115 return None

1116

1117 def pre_exec(self):

1118 super().pre_exec()

1119 if not getattr(self.compiled, "_oracle_cx_sql_compiler", False):

1120 return

1121

1122 self.out_parameters = {}

1123

1124 self._generate_out_parameter_vars()

1125

1126 self._generate_cursor_outputtype_handler()

1127

1128 def post_exec(self):

1129 if (

1130 self.compiled

1131 and is_sql_compiler(self.compiled)

1132 and self.compiled._oracle_returning

1133 ):

1134 initial_buffer = self.fetchall_for_returning(

1135 self.cursor, _internal=True

1136 )

1137

1138 fetch_strategy = _cursor.FullyBufferedCursorFetchStrategy(

1139 self.cursor,

1140 [

1141 (entry.keyname, None)

1142 for entry in self.compiled._result_columns

1143 ],

1144 initial_buffer=initial_buffer,

1145 )

1146

1147 self.cursor_fetch_strategy = fetch_strategy

1148

1149 def create_cursor(self):

1150 c = self._dbapi_connection.cursor()

1151 if self.dialect.arraysize:

1152 c.arraysize = self.dialect.arraysize

1153

1154 return c

1155

1156 def fetchall_for_returning(self, cursor, *, _internal=False):

1157 compiled = self.compiled

1158 if (

1159 not _internal

1160 and compiled is None

1161 or not is_sql_compiler(compiled)

1162 or not compiled._oracle_returning

1163 ):

1164 raise NotImplementedError(

1165 "execution context was not prepared for Oracle RETURNING"

1166 )

1167

1168 # create a fake cursor result from the out parameters. unlike

1169 # get_out_parameter_values(), the result-row handlers here will be

1170 # applied at the Result level

1171

1172 numcols = len(self.out_parameters)

1173

1174 # [stmt_result for stmt_result in outparam.values] == each

1175 # statement in executemany

1176 # [val for val in stmt_result] == each row for a particular

1177 # statement

1178 return list(

1179 zip(

1180 *[

1181 [

1182 val

1183 for stmt_result in self.out_parameters[

1184 f"ret_{j}"

1185 ].values

1186 for val in (stmt_result or ())

1187 ]

1188 for j in range(numcols)

1189 ]

1190 )

1191 )

1192

1193 def get_out_parameter_values(self, out_param_names):

1194 # this method should not be called when the compiler has

1195 # RETURNING as we've turned the has_out_parameters flag set to

1196 # False.

1197 assert not self.compiled.returning

1198

1199 return [

1200 self.dialect._paramval(self.out_parameters[name])

1201 for name in out_param_names

1202 ]

1203

1204

1205class OracleDialect_cx_oracle(OracleDialect):

1206 supports_statement_cache = True

1207 execution_ctx_cls = OracleExecutionContext_cx_oracle

1208 statement_compiler = OracleCompiler_cx_oracle

1209

1210 supports_sane_rowcount = True

1211 supports_sane_multi_rowcount = True

1212

1213 insert_executemany_returning = True

1214 insert_executemany_returning_sort_by_parameter_order = True

1215 update_executemany_returning = True

1216 delete_executemany_returning = True

1217

1218 supports_native_json_serialization = False

1219 supports_native_json_deserialization = False

1220 dialect_injects_custom_json_deserializer = True

1221

1222 bind_typing = interfaces.BindTyping.SETINPUTSIZES

1223

1224 minimum_dbapi_version = util.VersionInfo((8,))

1225

1226 driver = "cx_oracle"

1227

1228 colspecs = util.update_copy(

1229 OracleDialect.colspecs,

1230 {

1231 sqltypes.TIMESTAMP: _CXOracleTIMESTAMP,

1232 sqltypes.Numeric: _OracleNumeric,

1233 sqltypes.Float: _OracleFloat,

1234 oracle.BINARY_FLOAT: _OracleBINARY_FLOAT,

1235 oracle.BINARY_DOUBLE: _OracleBINARY_DOUBLE,

1236 sqltypes.JSON: _OracleJson,

1237 sqltypes.Integer: _OracleInteger,

1238 oracle.NUMBER: _OracleNUMBER,

1239 sqltypes.Date: _CXOracleDate,

1240 sqltypes.LargeBinary: _OracleBinary,

1241 sqltypes.Boolean: oracle._OracleBoolean,

1242 sqltypes.Interval: _OracleInterval,

1243 oracle.INTERVAL: _OracleInterval,

1244 sqltypes.Text: _OracleText,

1245 sqltypes.String: _OracleString,

1246 sqltypes.UnicodeText: _OracleUnicodeTextCLOB,

1247 sqltypes.CHAR: _OracleChar,

1248 sqltypes.NCHAR: _OracleNChar,

1249 sqltypes.Enum: _OracleEnum,

1250 oracle.LONG: _OracleLong,

1251 oracle.RAW: _OracleRaw,

1252 sqltypes.Unicode: _OracleUnicodeStringCHAR,

1253 sqltypes.NVARCHAR: _OracleUnicodeStringNCHAR,

1254 sqltypes.Uuid: _OracleUUID,

1255 oracle.NCLOB: _OracleUnicodeTextNCLOB,

1256 oracle.ROWID: _OracleRowid,

1257 },

1258 )

1259

1260 execute_sequence_format = list

1261

1262 _cursor_var_unicode_kwargs = util.immutabledict()

1263

1264 def __init__(

1265 self,

1266 auto_convert_lobs=True,

1267 coerce_to_decimal=True,

1268 arraysize=None,

1269 encoding_errors=None,

1270 **kwargs,

1271 ):

1272 OracleDialect.__init__(self, **kwargs)

1273 self.arraysize = arraysize

1274 self.encoding_errors = encoding_errors

1275 if encoding_errors:

1276 self._cursor_var_unicode_kwargs = {

1277 "encodingErrors": encoding_errors

1278 }

1279 self.auto_convert_lobs = auto_convert_lobs

1280 self.coerce_to_decimal = coerce_to_decimal

1281 if self._use_nchar_for_unicode:

1282 self.colspecs = self.colspecs.copy()

1283 self.colspecs[sqltypes.Unicode] = _OracleUnicodeStringNCHAR

1284 self.colspecs[sqltypes.UnicodeText] = _OracleUnicodeTextNCLOB

1285

1286 dbapi_module = self.dbapi

1287

1288 if dbapi_module is not None:

1289 # these constants will first be seen in SQLAlchemy datatypes

1290 # coming from the get_dbapi_type() method. We then

1291 # will place the following types into setinputsizes() calls

1292 # on each statement. Oracle constants that are not in this

1293 # list will not be put into setinputsizes().

1294 self.include_set_input_sizes = {

1295 dbapi_module.DATETIME,

1296 dbapi_module.DB_TYPE_NVARCHAR, # used for CLOB, NCLOB

1297 dbapi_module.DB_TYPE_RAW, # used for BLOB

1298 dbapi_module.NCLOB, # not currently used except for OUT param

1299 dbapi_module.CLOB, # not currently used except for OUT param

1300 dbapi_module.LOB, # not currently used

1301 dbapi_module.BLOB, # not currently used except for OUT param

1302 dbapi_module.NCHAR,

1303 dbapi_module.FIXED_NCHAR,

1304 dbapi_module.FIXED_CHAR,

1305 dbapi_module.TIMESTAMP,

1306 # we dont make use of Oracle's JSON serialization; does not

1307 # handle "none as null"

1308 # dbapi_module.DB_TYPE_JSON,

1309 int, # _OracleInteger,

1310 # _OracleBINARY_FLOAT, _OracleBINARY_DOUBLE,

1311 dbapi_module.NATIVE_FLOAT,

1312 }

1313

1314 self._paramval = lambda value: value.getvalue()

1315

1316 def retrieve_dbapi_version(self, dbapi):

1317 return util.parse_version_string(getattr(dbapi, "version", None))

1318

1319 @property

1320 def cx_oracle_ver(self):

1321 """Legacy accessor for :attr:`.Dialect.dbapi_version`.

1322

1323 Retained for backwards compatibility; ``(0, 0, 0)`` is returned

1324 when no version can be determined.

1325

1326 """

1327 version = self._dbapi_version_or_none

1328 return version if version is not None else util.VersionInfo((0, 0, 0))

1329

1330 @classmethod

1331 def import_dbapi(cls):

1332 import cx_Oracle

1333

1334 return cx_Oracle

1335

1336 def initialize(self, connection):

1337 super().initialize(connection)

1338 self._detect_decimal_char(connection)

1339

1340 def get_isolation_level(self, dbapi_connection):

1341 # sources:

1342

1343 # general idea of transaction id, have to start one, etc.

1344 # https://stackoverflow.com/questions/10711204/how-to-check-isoloation-level

1345

1346 # how to decode xid cols from v$transaction to match

1347 # https://asktom.oracle.com/pls/apex/f?p=100:11:0::::P11_QUESTION_ID:9532779900346079444

1348

1349 # Oracle tuple comparison without using IN:

1350 # https://www.sql-workbench.eu/comparison/tuple_comparison.html

1351

1352 with dbapi_connection.cursor() as cursor:

1353 # this is the only way to ensure a transaction is started without

1354 # actually running DML. There's no way to see the configured

1355 # isolation level without getting it from v$transaction which

1356 # means transaction has to be started.

1357 outval = cursor.var(str)

1358 cursor.execute(

1359 """

1360 begin

1361 :trans_id := dbms_transaction.local_transaction_id( TRUE );

1362 end;

1363 """,

1364 {"trans_id": outval},

1365 )

1366 trans_id = outval.getvalue()

1367 xidusn, xidslot, xidsqn = trans_id.split(".", 2)

1368

1369 cursor.execute(

1370 "SELECT CASE BITAND(t.flag, POWER(2, 28)) "

1371 "WHEN 0 THEN 'READ COMMITTED' "

1372 "ELSE 'SERIALIZABLE' END AS isolation_level "

1373 "FROM v$transaction t WHERE "

1374 "(t.xidusn, t.xidslot, t.xidsqn) = "

1375 "((:xidusn, :xidslot, :xidsqn))",

1376 {"xidusn": xidusn, "xidslot": xidslot, "xidsqn": xidsqn},

1377 )

1378 row = cursor.fetchone()

1379 if row is None:

1380 raise exc.InvalidRequestError(

1381 "could not retrieve isolation level"

1382 )

1383 result = row[0]

1384

1385 return result

1386

1387 def get_isolation_level_values(self, dbapi_connection):

1388 return super().get_isolation_level_values(dbapi_connection) + [

1389 "AUTOCOMMIT"

1390 ]

1391

1392 def set_isolation_level(self, dbapi_connection, level):

1393 if level == "AUTOCOMMIT":

1394 dbapi_connection.autocommit = True

1395 else:

1396 dbapi_connection.autocommit = False

1397 dbapi_connection.rollback()

1398 with dbapi_connection.cursor() as cursor:

1399 cursor.execute(f"ALTER SESSION SET ISOLATION_LEVEL={level}")

1400

1401 def detect_autocommit_setting(self, dbapi_conn) -> bool:

1402 return bool(dbapi_conn.autocommit)

1403

1404 def _detect_decimal_char(self, connection):

1405 # we have the option to change this setting upon connect,

1406 # or just look at what it is upon connect and convert.

1407 # to minimize the chance of interference with changes to

1408 # NLS_TERRITORY or formatting behavior of the DB, we opt

1409 # to just look at it

1410

1411 dbapi_connection = connection.connection

1412

1413 with dbapi_connection.cursor() as cursor:

1414 # issue #8744

1415 # nls_session_parameters is not available in some Oracle

1416 # modes like "mount mode". But then, v$nls_parameters is not

1417 # available if the connection doesn't have SYSDBA priv.

1418 #

1419 # simplify the whole thing and just use the method that we were

1420 # doing in the test suite already, selecting a number

1421

1422 def output_type_handler(

1423 cursor, name, defaultType, size, precision, scale

1424 ):

1425 return cursor.var(

1426 self.dbapi.STRING, 255, arraysize=cursor.arraysize

1427 )

1428

1429 cursor.outputtypehandler = output_type_handler

1430 cursor.execute("SELECT 1.1 FROM DUAL")

1431 value = cursor.fetchone()[0]

1432

1433 decimal_char = value.lstrip("0")[1]

1434 assert not decimal_char[0].isdigit()

1435

1436 self._decimal_char = decimal_char

1437

1438 if self._decimal_char != ".":

1439 _detect_decimal = self._detect_decimal

1440 _to_decimal = self._to_decimal

1441

1442 self._detect_decimal = lambda value: _detect_decimal(

1443 value.replace(self._decimal_char, ".")

1444 )

1445 self._to_decimal = lambda value: _to_decimal(

1446 value.replace(self._decimal_char, ".")

1447 )

1448

1449 def _detect_decimal(self, value):

1450 if "." in value:

1451 return self._to_decimal(value)

1452 else:

1453 return int(value)

1454

1455 _to_decimal = decimal.Decimal

1456

1457 def _cursor_lob_as_str_var(self, cursor, default_type, outconverter=None):

1458 """Produce a ``cursor.var()`` that receives a CLOB / NCLOB as a

1459 string.

1460

1461 Used both by the connection-wide outputtypehandler and by

1462 type-level handlers such as that of :class:`._OracleJson`, which

1463 need the LOB contents as a string in order to decode them further.

1464

1465 """

1466 cx_Oracle = self.dbapi

1467

1468 kw = self._cursor_var_unicode_kwargs

1469 if outconverter is not None:

1470 kw = {**kw, "outconverter": outconverter}

1471

1472 return cursor.var(

1473 (

1474 cx_Oracle.DB_TYPE_VARCHAR

1475 if default_type is cx_Oracle.CLOB

1476 else cx_Oracle.DB_TYPE_NVARCHAR

1477 ),

1478 _CX_ORACLE_MAGIC_LOB_SIZE,

1479 cursor.arraysize,

1480 **kw,

1481 )

1482

1483 def _generate_connection_outputtype_handler(self):

1484 """establish the default outputtypehandler established at the

1485 connection level.

1486

1487 note that when using a Compiled statement that has types (e.g.

1488 TypeEngine), we set up a per-cursor handler instead which supercedes

1489 this one, using Oracle-specific TypeEngine handlers delivered by the

1490 _cx_oracle_outputtypehandler() method of each one.

1491

1492 """

1493

1494 dialect = self

1495 cx_Oracle = dialect.dbapi

1496

1497 number_handler = _OracleNUMBER(

1498 asdecimal=True

1499 )._cx_oracle_outputtypehandler(dialect)

1500 float_handler = _OracleNUMBER(

1501 asdecimal=False

1502 )._cx_oracle_outputtypehandler(dialect)

1503

1504 def output_type_handler(

1505 cursor, name, default_type, size, precision, scale

1506 ):

1507 if (

1508 default_type == cx_Oracle.NUMBER

1509 and default_type is not cx_Oracle.NATIVE_FLOAT

1510 ):

1511 if not dialect.coerce_to_decimal:

1512 return None

1513 elif precision == 0 and scale in (0, -127):

1514 # ambiguous type, this occurs when selecting

1515 # numbers from deep subqueries

1516 return cursor.var(

1517 cx_Oracle.STRING,

1518 255,

1519 outconverter=dialect._detect_decimal,

1520 arraysize=cursor.arraysize,

1521 )

1522 elif precision and scale > 0:

1523 return number_handler(

1524 cursor, name, default_type, size, precision, scale

1525 )

1526 else:

1527 return float_handler(

1528 cursor, name, default_type, size, precision, scale

1529 )

1530

1531 # if unicode options were specified, add a decoder, otherwise

1532 # cx_Oracle should return Unicode

1533 elif (

1534 dialect._cursor_var_unicode_kwargs

1535 and default_type

1536 in (

1537 cx_Oracle.STRING,

1538 cx_Oracle.FIXED_CHAR,

1539 )

1540 and default_type is not cx_Oracle.CLOB

1541 and default_type is not cx_Oracle.NCLOB

1542 ):

1543 return cursor.var(

1544 str,

1545 size,

1546 cursor.arraysize,

1547 **dialect._cursor_var_unicode_kwargs,

1548 )

1549

1550 elif dialect.auto_convert_lobs and default_type in (

1551 cx_Oracle.CLOB,

1552 cx_Oracle.NCLOB,

1553 ):

1554 return dialect._cursor_lob_as_str_var(cursor, default_type)

1555

1556 elif dialect.auto_convert_lobs and default_type in (

1557 cx_Oracle.BLOB,

1558 ):

1559 return cursor.var(

1560 cx_Oracle.DB_TYPE_RAW,

1561 _CX_ORACLE_MAGIC_LOB_SIZE,

1562 cursor.arraysize,

1563 )

1564 elif (

1565 default_type is cx_Oracle.DB_TYPE_JSON

1566 and dialect._json_deserializer is not None

1567 ):

1568 return cursor.var(

1569 cx_Oracle.DB_TYPE_VARCHAR,

1570 _CX_ORACLE_MAX_JSON_CONVERTED,

1571 cursor.arraysize,

1572 outconverter=dialect._json_deserializer,

1573 )

1574

1575 return output_type_handler

1576

1577 def on_connect(self):

1578 output_type_handler = self._generate_connection_outputtype_handler()

1579

1580 def on_connect(conn):

1581 conn.outputtypehandler = output_type_handler

1582

1583 return on_connect

1584

1585 def create_connect_args(self, url):

1586 opts = dict(url.query)

1587

1588 database = url.database

1589 service_name = opts.pop("service_name", None)

1590 if database or service_name:

1591 # if we have a database, then we have a remote host

1592 port = url.port

1593 if port:

1594 port = int(port)

1595 else:

1596 port = 1521

1597

1598 if database and service_name:

1599 raise exc.InvalidRequestError(

1600 '"service_name" option shouldn\'t '

1601 'be used with a "database" part of the url'

1602 )

1603 if database:

1604 makedsn_kwargs = {"sid": database}

1605 if service_name:

1606 makedsn_kwargs = {"service_name": service_name}

1607

1608 dsn = self.dbapi.makedsn(url.host, port, **makedsn_kwargs)

1609 else:

1610 # we have a local tnsname

1611 dsn = url.host

1612

1613 if dsn is not None:

1614 opts["dsn"] = dsn

1615 if url.password is not None:

1616 opts["password"] = url.password

1617 if url.username is not None:

1618 opts["user"] = url.username

1619

1620 def convert_cx_oracle_constant(value):

1621 if isinstance(value, str):

1622 try:

1623 int_val = int(value)

1624 except ValueError:

1625 value = value.upper()

1626 return getattr(self.dbapi, value)

1627 else:

1628 return int_val

1629 else:

1630 return value

1631

1632 util.coerce_kw_type(opts, "mode", convert_cx_oracle_constant)

1633 util.coerce_kw_type(opts, "threaded", bool)

1634 util.coerce_kw_type(opts, "events", bool)

1635 util.coerce_kw_type(opts, "purity", convert_cx_oracle_constant)

1636 return ([], opts)

1637

1638 def _get_server_version_info(self, connection):

1639 return tuple(int(x) for x in connection.connection.version.split("."))

1640

1641 def is_disconnect(self, e, connection, cursor):

1642 (error,) = e.args

1643 if isinstance(

1644 e, (self.dbapi.InterfaceError, self.dbapi.DatabaseError)

1645 ) and "not connected" in str(e):

1646 return True

1647

1648 if hasattr(error, "code") and error.code in {

1649 28,

1650 3114,

1651 3113,

1652 3135,

1653 1033,

1654 2396,

1655 }:

1656 # ORA-00028: your session has been killed

1657 # ORA-03114: not connected to ORACLE

1658 # ORA-03113: end-of-file on communication channel

1659 # ORA-03135: connection lost contact

1660 # ORA-01033: ORACLE initialization or shutdown in progress

1661 # ORA-02396: exceeded maximum idle time, please connect again

1662 # TODO: Others ?

1663 return True

1664

1665 if re.match(r"^(?:DPI-1010|DPI-1080|DPY-1001|DPY-4011)", str(e)):

1666 # DPI-1010: not connected

1667 # DPI-1080: connection was closed by ORA-3113

1668 # python-oracledb's DPY-1001: not connected to database

1669 # python-oracledb's DPY-4011: the database or network closed the

1670 # connection

1671 # TODO: others?

1672 return True

1673

1674 return False

1675

1676 def create_xid(self):

1677 id_ = random.randint(0, 2**128)

1678 return (0x1234, "%032x" % id_, "%032x" % 9)

1679

1680 def do_executemany(self, cursor, statement, parameters, context=None):

1681 if isinstance(parameters, tuple):

1682 parameters = list(parameters)

1683 cursor.executemany(statement, parameters)

1684

1685 def do_begin_twophase(self, connection, xid):

1686 connection.connection.begin(*xid)

1687 connection.connection.info["cx_oracle_xid"] = xid

1688

1689 def do_prepare_twophase(self, connection, xid):

1690 result = connection.connection.prepare()

1691 connection.info["cx_oracle_prepared"] = result

1692

1693 def do_rollback_twophase(

1694 self, connection, xid, is_prepared=True, recover=False

1695 ):

1696 self.do_rollback(connection.connection)

1697 # TODO: need to end XA state here

1698

1699 def do_commit_twophase(

1700 self, connection, xid, is_prepared=True, recover=False

1701 ):

1702 if not is_prepared:

1703 self.do_commit(connection.connection)

1704 else:

1705 if recover:

1706 raise NotImplementedError(

1707 "2pc recovery not implemented for cx_Oracle"

1708 )

1709 oci_prepared = connection.info["cx_oracle_prepared"]

1710 if oci_prepared:

1711 self.do_commit(connection.connection)

1712 # TODO: need to end XA state here

1713

1714 def do_set_input_sizes(self, cursor, list_of_tuples, context):

1715 if self.positional:

1716 # not usually used, here to support if someone is modifying

1717 # the dialect to use positional style

1718 cursor.setinputsizes(

1719 *[dbtype for key, dbtype, sqltype in list_of_tuples]

1720 )

1721 else:

1722 collection = (

1723 (key, dbtype)

1724 for key, dbtype, sqltype in list_of_tuples

1725 if dbtype

1726 )

1727

1728 cursor.setinputsizes(**{key: dbtype for key, dbtype in collection})

1729

1730 def do_recover_twophase(self, connection):

1731 raise NotImplementedError(

1732 "recover two phase query for cx_Oracle not implemented"

1733 )

1734

1735

1736dialect = OracleDialect_cx_oracle

Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/dialects/oracle/cx_oracle.py: 30%

512 statements