Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# dialects/oracle/cx_oracle.py
2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
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)))'
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.
109Passing cx_Oracle connect arguments
110-----------------------------------
112Additional connection arguments can usually be passed via the URL query string;
113particular symbols like ``SYSDBA`` are intercepted and converted to the correct
114symbol::
116 e = create_engine(
117 "oracle+cx_oracle://user:pass@dsn?encoding=UTF-8&nencoding=UTF-8&mode=SYSDBA&events=true"
118 )
120.. versionchanged:: 1.3 the cx_Oracle dialect now accepts all argument names
121 within the URL string itself, to be passed to the cx_Oracle DBAPI. As
122 was the case earlier but not correctly documented, the
123 :paramref:`_sa.create_engine.connect_args` parameter also accepts all
124 cx_Oracle DBAPI connect arguments.
126To pass arguments directly to ``.connect()`` without using the query
127string, use the :paramref:`_sa.create_engine.connect_args` dictionary.
128Any cx_Oracle parameter value and/or constant may be passed, such as::
130 import cx_Oracle
132 e = create_engine(
133 "oracle+cx_oracle://user:pass@dsn",
134 connect_args={
135 "encoding": "UTF-8",
136 "nencoding": "UTF-8",
137 "mode": cx_Oracle.SYSDBA,
138 "events": True,
139 },
140 )
142Note that the default driver value for ``encoding`` and ``nencoding`` was
143changed to "UTF-8" in cx_Oracle 8.0 so these parameters can be omitted when
144using that version, or later.
146Options consumed by the SQLAlchemy cx_Oracle dialect outside of the driver
147--------------------------------------------------------------------------
149There are also options that are consumed by the SQLAlchemy cx_oracle dialect
150itself. These options are always passed directly to :func:`_sa.create_engine`
151, such as::
153 e = create_engine(
154 "oracle+cx_oracle://user:pass@dsn", coerce_to_decimal=False
155 )
157The parameters accepted by the cx_oracle dialect are as follows:
159* ``arraysize`` - set the cx_oracle.arraysize value on cursors; defaults
160 to ``None``, indicating that the driver default should be used (typically
161 the value is 100). This setting controls how many rows are buffered when
162 fetching rows, and can have a significant effect on performance when
163 modified.
165 .. versionchanged:: 2.0.26 - changed the default value from 50 to None,
166 to use the default value of the driver itself.
168* ``auto_convert_lobs`` - defaults to True; See :ref:`cx_oracle_lob`.
170* ``coerce_to_decimal`` - see :ref:`cx_oracle_numeric` for detail.
172* ``encoding_errors`` - see :ref:`cx_oracle_unicode_encoding_errors` for detail.
174.. _cx_oracle_sessionpool:
176Using cx_Oracle SessionPool
177---------------------------
179The cx_Oracle driver provides its own connection pool implementation that may
180be used in place of SQLAlchemy's pooling functionality. The driver pool
181supports Oracle Database features such dead connection detection, connection
182draining for planned database downtime, support for Oracle Application
183Continuity and Transparent Application Continuity, and gives support for
184Database Resident Connection Pooling (DRCP).
186Using the driver pool can be achieved by using the
187:paramref:`_sa.create_engine.creator` parameter to provide a function that
188returns a new connection, along with setting
189:paramref:`_sa.create_engine.pool_class` to ``NullPool`` to disable
190SQLAlchemy's pooling::
192 import cx_Oracle
193 from sqlalchemy import create_engine
194 from sqlalchemy.pool import NullPool
196 pool = cx_Oracle.SessionPool(
197 user="scott",
198 password="tiger",
199 dsn="orclpdb",
200 min=1,
201 max=4,
202 increment=1,
203 threaded=True,
204 encoding="UTF-8",
205 nencoding="UTF-8",
206 )
208 engine = create_engine(
209 "oracle+cx_oracle://", creator=pool.acquire, poolclass=NullPool
210 )
212The above engine may then be used normally where cx_Oracle's pool handles
213connection pooling::
215 with engine.connect() as conn:
216 print(conn.scalar("select 1 from dual"))
218As well as providing a scalable solution for multi-user applications, the
219cx_Oracle session pool supports some Oracle features such as DRCP and
220`Application Continuity
221<https://cx-oracle.readthedocs.io/en/latest/user_guide/ha.html#application-continuity-ac>`_.
223Note that the pool creation parameters ``threaded``, ``encoding`` and
224``nencoding`` were deprecated in later cx_Oracle releases.
226Using Oracle Database Resident Connection Pooling (DRCP)
227--------------------------------------------------------
229When using Oracle Database's DRCP, the best practice is to pass a connection
230class and "purity" when acquiring a connection from the SessionPool. Refer to
231the `cx_Oracle DRCP documentation
232<https://cx-oracle.readthedocs.io/en/latest/user_guide/connection_handling.html#database-resident-connection-pooling-drcp>`_.
234This can be achieved by wrapping ``pool.acquire()``::
236 import cx_Oracle
237 from sqlalchemy import create_engine
238 from sqlalchemy.pool import NullPool
240 pool = cx_Oracle.SessionPool(
241 user="scott",
242 password="tiger",
243 dsn="orclpdb",
244 min=2,
245 max=5,
246 increment=1,
247 threaded=True,
248 encoding="UTF-8",
249 nencoding="UTF-8",
250 )
253 def creator():
254 return pool.acquire(
255 cclass="MYCLASS", purity=cx_Oracle.ATTR_PURITY_SELF
256 )
259 engine = create_engine(
260 "oracle+cx_oracle://", creator=creator, poolclass=NullPool
261 )
263The above engine may then be used normally where cx_Oracle handles session
264pooling and Oracle Database additionally uses DRCP::
266 with engine.connect() as conn:
267 print(conn.scalar("select 1 from dual"))
269.. _cx_oracle_unicode:
271Unicode
272-------
274As is the case for all DBAPIs under Python 3, all strings are inherently
275Unicode strings. In all cases however, the driver requires an explicit
276encoding configuration.
278Ensuring the Correct Client Encoding
279^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
281The long accepted standard for establishing client encoding for nearly all
282Oracle Database related software is via the `NLS_LANG
283<https://www.oracle.com/database/technologies/faq-nls-lang.html>`_ environment
284variable. Older versions of cx_Oracle use this environment variable as the
285source of its encoding configuration. The format of this variable is
286Territory_Country.CharacterSet; a typical value would be
287``AMERICAN_AMERICA.AL32UTF8``. cx_Oracle version 8 and later use the character
288set "UTF-8" by default, and ignore the character set component of NLS_LANG.
290The cx_Oracle driver also supported a programmatic alternative which is to pass
291the ``encoding`` and ``nencoding`` parameters directly to its ``.connect()``
292function. These can be present in the URL as follows::
294 engine = create_engine(
295 "oracle+cx_oracle://scott:tiger@tnsalias?encoding=UTF-8&nencoding=UTF-8"
296 )
298For the meaning of the ``encoding`` and ``nencoding`` parameters, please
299consult
300`Characters Sets and National Language Support (NLS) <https://cx-oracle.readthedocs.io/en/latest/user_guide/globalization.html#globalization>`_.
302.. seealso::
304 `Characters Sets and National Language Support (NLS) <https://cx-oracle.readthedocs.io/en/latest/user_guide/globalization.html#globalization>`_
305 - in the cx_Oracle documentation.
308Unicode-specific Column datatypes
309^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
311The Core expression language handles unicode data by use of the
312:class:`.Unicode` and :class:`.UnicodeText` datatypes. These types correspond
313to the VARCHAR2 and CLOB Oracle Database datatypes by default. When using
314these datatypes with Unicode data, it is expected that the database is
315configured with a Unicode-aware character set, as well as that the ``NLS_LANG``
316environment variable is set appropriately (this applies to older versions of
317cx_Oracle), so that the VARCHAR2 and CLOB datatypes can accommodate the data.
319In the case that Oracle Database is not configured with a Unicode character
320set, the two options are to use the :class:`_types.NCHAR` and
321:class:`_oracle.NCLOB` datatypes explicitly, or to pass the flag
322``use_nchar_for_unicode=True`` to :func:`_sa.create_engine`, which will cause
323the SQLAlchemy dialect to use NCHAR/NCLOB for the :class:`.Unicode` /
324:class:`.UnicodeText` datatypes instead of VARCHAR/CLOB.
326.. versionchanged:: 1.3 The :class:`.Unicode` and :class:`.UnicodeText`
327 datatypes now correspond to the ``VARCHAR2`` and ``CLOB`` Oracle Database
328 datatypes unless the ``use_nchar_for_unicode=True`` is passed to the dialect
329 when :func:`_sa.create_engine` is called.
332.. _cx_oracle_unicode_encoding_errors:
334Encoding Errors
335^^^^^^^^^^^^^^^
337For the unusual case that data in Oracle Database is present with a broken
338encoding, the dialect accepts a parameter ``encoding_errors`` which will be
339passed to Unicode decoding functions in order to affect how decoding errors are
340handled. The value is ultimately consumed by the Python `decode
341<https://docs.python.org/3/library/stdtypes.html#bytes.decode>`_ function, and
342is passed both via cx_Oracle's ``encodingErrors`` parameter consumed by
343``Cursor.var()``, as well as SQLAlchemy's own decoding function, as the
344cx_Oracle dialect makes use of both under different circumstances.
346.. versionadded:: 1.3.11
349.. _cx_oracle_setinputsizes:
351Fine grained control over cx_Oracle data binding performance with setinputsizes
352-------------------------------------------------------------------------------
354The cx_Oracle DBAPI has a deep and fundamental reliance upon the usage of the
355DBAPI ``setinputsizes()`` call. The purpose of this call is to establish the
356datatypes that are bound to a SQL statement for Python values being passed as
357parameters. While virtually no other DBAPI assigns any use to the
358``setinputsizes()`` call, the cx_Oracle DBAPI relies upon it heavily in its
359interactions with the Oracle Database client interface, and in some scenarios
360it is not possible for SQLAlchemy to know exactly how data should be bound, as
361some settings can cause profoundly different performance characteristics, while
362altering the type coercion behavior at the same time.
364Users of the cx_Oracle dialect are **strongly encouraged** to read through
365cx_Oracle's list of built-in datatype symbols at
366https://cx-oracle.readthedocs.io/en/latest/api_manual/module.html#database-types.
367Note that in some cases, significant performance degradation can occur when
368using these types vs. not, in particular when specifying ``cx_Oracle.CLOB``.
370On the SQLAlchemy side, the :meth:`.DialectEvents.do_setinputsizes` event can
371be used both for runtime visibility (e.g. logging) of the setinputsizes step as
372well as to fully control how ``setinputsizes()`` is used on a per-statement
373basis.
375.. versionadded:: 1.2.9 Added :meth:`.DialectEvents.setinputsizes`
378Example 1 - logging all setinputsizes calls
379^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
381The following example illustrates how to log the intermediary values from a
382SQLAlchemy perspective before they are converted to the raw ``setinputsizes()``
383parameter dictionary. The keys of the dictionary are :class:`.BindParameter`
384objects which have a ``.key`` and a ``.type`` attribute::
386 from sqlalchemy import create_engine, event
388 engine = create_engine("oracle+cx_oracle://scott:tiger@host/xe")
391 @event.listens_for(engine, "do_setinputsizes")
392 def _log_setinputsizes(inputsizes, cursor, statement, parameters, context):
393 for bindparam, dbapitype in inputsizes.items():
394 log.info(
395 "Bound parameter name: %s SQLAlchemy type: %r DBAPI object: %s",
396 bindparam.key,
397 bindparam.type,
398 dbapitype,
399 )
401Example 2 - remove all bindings to CLOB
402^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
404The ``CLOB`` datatype in cx_Oracle incurs a significant performance overhead,
405however is set by default for the ``Text`` type within the SQLAlchemy 1.2
406series. This setting can be modified as follows::
408 from sqlalchemy import create_engine, event
409 from cx_Oracle import CLOB
411 engine = create_engine("oracle+cx_oracle://scott:tiger@host/xe")
414 @event.listens_for(engine, "do_setinputsizes")
415 def _remove_clob(inputsizes, cursor, statement, parameters, context):
416 for bindparam, dbapitype in list(inputsizes.items()):
417 if dbapitype is CLOB:
418 del inputsizes[bindparam]
420.. _cx_oracle_lob:
422LOB Datatypes
423--------------
425LOB datatypes refer to the "large object" datatypes such as CLOB, NCLOB and
426BLOB. Modern versions of cx_Oracle is optimized for these datatypes to be
427delivered as a single buffer. As such, SQLAlchemy makes use of these newer type
428handlers by default.
430To disable the use of newer type handlers and deliver LOB objects as classic
431buffered objects with a ``read()`` method, the parameter
432``auto_convert_lobs=False`` may be passed to :func:`_sa.create_engine`,
433which takes place only engine-wide.
435.. _cx_oracle_returning:
437RETURNING Support
438-----------------
440The cx_Oracle dialect implements RETURNING using OUT parameters.
441The dialect supports RETURNING fully.
443Two Phase Transactions Not Supported
444------------------------------------
446Two phase transactions are **not supported** under cx_Oracle due to poor driver
447support. The newer :ref:`oracledb` dialect however **does** support two phase
448transactions.
450.. _cx_oracle_numeric:
452Precision Numerics
453------------------
455SQLAlchemy's numeric types can handle receiving and returning values as Python
456``Decimal`` objects or float objects. When a :class:`.Numeric` object, or a
457subclass such as :class:`.Float`, :class:`_oracle.DOUBLE_PRECISION` etc. is in
458use, the :paramref:`.Numeric.asdecimal` flag determines if values should be
459coerced to ``Decimal`` upon return, or returned as float objects. To make
460matters more complicated under Oracle Database, the ``NUMBER`` type can also
461represent integer values if the "scale" is zero, so the Oracle
462Database-specific :class:`_oracle.NUMBER` type takes this into account as well.
464The cx_Oracle dialect makes extensive use of connection- and cursor-level
465"outputtypehandler" callables in order to coerce numeric values as requested.
466These callables are specific to the specific flavor of :class:`.Numeric` in
467use, as well as if no SQLAlchemy typing objects are present. There are
468observed scenarios where Oracle Database may send incomplete or ambiguous
469information about the numeric types being returned, such as a query where the
470numeric types are buried under multiple levels of subquery. The type handlers
471do their best to make the right decision in all cases, deferring to the
472underlying cx_Oracle DBAPI for all those cases where the driver can make the
473best decision.
475When no typing objects are present, as when executing plain SQL strings, a
476default "outputtypehandler" is present which will generally return numeric
477values which specify precision and scale as Python ``Decimal`` objects. To
478disable this coercion to decimal for performance reasons, pass the flag
479``coerce_to_decimal=False`` to :func:`_sa.create_engine`::
481 engine = create_engine("oracle+cx_oracle://dsn", coerce_to_decimal=False)
483The ``coerce_to_decimal`` flag only impacts the results of plain string
484SQL statements that are not otherwise associated with a :class:`.Numeric`
485SQLAlchemy type (or a subclass of such).
487.. versionchanged:: 1.2 The numeric handling system for cx_Oracle has been
488 reworked to take advantage of newer cx_Oracle features as well
489 as better integration of outputtypehandlers.
491""" # noqa
493from __future__ import annotations
495import decimal
496import random
497import re
499from . import base as oracle
500from .base import OracleCompiler
501from .base import OracleDialect
502from .base import OracleExecutionContext
503from .types import _OracleDateLiteralRender
504from ... import exc
505from ... import util
506from ...engine import cursor as _cursor
507from ...engine import interfaces
508from ...engine import processors
509from ...sql import sqltypes
510from ...sql._typing import is_sql_compiler
512# source:
513# https://github.com/oracle/python-cx_Oracle/issues/596#issuecomment-999243649
514_CX_ORACLE_MAGIC_LOB_SIZE = 131072
517class _OracleInteger(sqltypes.Integer):
518 def get_dbapi_type(self, dbapi):
519 # see https://github.com/oracle/python-cx_Oracle/issues/
520 # 208#issuecomment-409715955
521 return int
523 def _cx_oracle_var(self, dialect, cursor, arraysize=None):
524 cx_Oracle = dialect.dbapi
525 return cursor.var(
526 cx_Oracle.STRING,
527 255,
528 arraysize=arraysize if arraysize is not None else cursor.arraysize,
529 outconverter=int,
530 )
532 def _cx_oracle_outputtypehandler(self, dialect):
533 def handler(cursor, name, default_type, size, precision, scale):
534 return self._cx_oracle_var(dialect, cursor)
536 return handler
539class _OracleNumeric(sqltypes.Numeric):
540 is_number = False
542 def bind_processor(self, dialect):
543 if self.scale == 0:
544 return None
545 elif self.asdecimal:
546 processor = processors.to_decimal_processor_factory(
547 decimal.Decimal, self._effective_decimal_return_scale
548 )
550 def process(value):
551 if isinstance(value, (int, float)):
552 return processor(value)
553 elif value is not None and value.is_infinite():
554 return float(value)
555 else:
556 return value
558 return process
559 else:
560 return processors.to_float
562 def result_processor(self, dialect, coltype):
563 return None
565 def _cx_oracle_outputtypehandler(self, dialect):
566 cx_Oracle = dialect.dbapi
568 def handler(cursor, name, default_type, size, precision, scale):
569 outconverter = None
571 if precision:
572 if self.asdecimal:
573 if default_type == cx_Oracle.NATIVE_FLOAT:
574 # receiving float and doing Decimal after the fact
575 # allows for float("inf") to be handled
576 type_ = default_type
577 outconverter = decimal.Decimal
578 else:
579 type_ = decimal.Decimal
580 else:
581 if self.is_number and scale == 0:
582 # integer. cx_Oracle is observed to handle the widest
583 # variety of ints when no directives are passed,
584 # from 5.2 to 7.0. See [ticket:4457]
585 return None
586 else:
587 type_ = cx_Oracle.NATIVE_FLOAT
589 else:
590 if self.asdecimal:
591 if default_type == cx_Oracle.NATIVE_FLOAT:
592 type_ = default_type
593 outconverter = decimal.Decimal
594 else:
595 type_ = decimal.Decimal
596 else:
597 if self.is_number and scale == 0:
598 # integer. cx_Oracle is observed to handle the widest
599 # variety of ints when no directives are passed,
600 # from 5.2 to 7.0. See [ticket:4457]
601 return None
602 else:
603 type_ = cx_Oracle.NATIVE_FLOAT
605 return cursor.var(
606 type_,
607 255,
608 arraysize=cursor.arraysize,
609 outconverter=outconverter,
610 )
612 return handler
615class _OracleUUID(sqltypes.Uuid):
616 def get_dbapi_type(self, dbapi):
617 return dbapi.STRING
620class _OracleBinaryFloat(_OracleNumeric):
621 def get_dbapi_type(self, dbapi):
622 return dbapi.NATIVE_FLOAT
625class _OracleBINARY_FLOAT(_OracleBinaryFloat, oracle.BINARY_FLOAT):
626 pass
629class _OracleBINARY_DOUBLE(_OracleBinaryFloat, oracle.BINARY_DOUBLE):
630 pass
633class _OracleNUMBER(_OracleNumeric):
634 is_number = True
637class _CXOracleDate(oracle._OracleDate):
638 def bind_processor(self, dialect):
639 return None
641 def result_processor(self, dialect, coltype):
642 def process(value):
643 if value is not None:
644 return value.date()
645 else:
646 return value
648 return process
651class _CXOracleTIMESTAMP(_OracleDateLiteralRender, sqltypes.TIMESTAMP):
652 def literal_processor(self, dialect):
653 return self._literal_processor_datetime(dialect)
656class _LOBDataType:
657 pass
660# TODO: the names used across CHAR / VARCHAR / NCHAR / NVARCHAR
661# here are inconsistent and not very good
662class _OracleChar(sqltypes.CHAR):
663 def get_dbapi_type(self, dbapi):
664 return dbapi.FIXED_CHAR
667class _OracleNChar(sqltypes.NCHAR):
668 def get_dbapi_type(self, dbapi):
669 return dbapi.FIXED_NCHAR
672class _OracleUnicodeStringNCHAR(oracle.NVARCHAR2):
673 def get_dbapi_type(self, dbapi):
674 return dbapi.NCHAR
677class _OracleUnicodeStringCHAR(sqltypes.Unicode):
678 def get_dbapi_type(self, dbapi):
679 return dbapi.LONG_STRING
682class _OracleUnicodeTextNCLOB(_LOBDataType, oracle.NCLOB):
683 def get_dbapi_type(self, dbapi):
684 # previously, this was dbapi.NCLOB.
685 # DB_TYPE_NVARCHAR will instead be passed to setinputsizes()
686 # when this datatype is used.
687 return dbapi.DB_TYPE_NVARCHAR
690class _OracleUnicodeTextCLOB(_LOBDataType, sqltypes.UnicodeText):
691 def get_dbapi_type(self, dbapi):
692 # previously, this was dbapi.CLOB.
693 # DB_TYPE_NVARCHAR will instead be passed to setinputsizes()
694 # when this datatype is used.
695 return dbapi.DB_TYPE_NVARCHAR
698class _OracleText(_LOBDataType, sqltypes.Text):
699 def get_dbapi_type(self, dbapi):
700 # previously, this was dbapi.CLOB.
701 # DB_TYPE_NVARCHAR will instead be passed to setinputsizes()
702 # when this datatype is used.
703 return dbapi.DB_TYPE_NVARCHAR
706class _OracleLong(_LOBDataType, oracle.LONG):
707 def get_dbapi_type(self, dbapi):
708 return dbapi.LONG_STRING
711class _OracleString(sqltypes.String):
712 pass
715class _OracleEnum(sqltypes.Enum):
716 def bind_processor(self, dialect):
717 enum_proc = sqltypes.Enum.bind_processor(self, dialect)
719 def process(value):
720 raw_str = enum_proc(value)
721 return raw_str
723 return process
726class _OracleBinary(_LOBDataType, sqltypes.LargeBinary):
727 def get_dbapi_type(self, dbapi):
728 # previously, this was dbapi.BLOB.
729 # DB_TYPE_RAW will instead be passed to setinputsizes()
730 # when this datatype is used.
731 return dbapi.DB_TYPE_RAW
733 def bind_processor(self, dialect):
734 return None
736 def result_processor(self, dialect, coltype):
737 if not dialect.auto_convert_lobs:
738 return None
739 else:
740 return super().result_processor(dialect, coltype)
743class _OracleInterval(oracle.INTERVAL):
744 def get_dbapi_type(self, dbapi):
745 return dbapi.INTERVAL
748class _OracleRaw(oracle.RAW):
749 pass
752class _OracleRowid(oracle.ROWID):
753 def get_dbapi_type(self, dbapi):
754 return dbapi.ROWID
757class OracleCompiler_cx_oracle(OracleCompiler):
758 _oracle_cx_sql_compiler = True
760 _oracle_returning = False
762 # Oracle bind names can't start with digits or underscores.
763 # currently we rely upon Oracle-specific quoting of bind names in most
764 # cases. however for expanding params, the escape chars are used.
765 # see #8708
766 bindname_escape_characters = util.immutabledict(
767 {
768 "%": "P",
769 "(": "A",
770 ")": "Z",
771 ":": "C",
772 ".": "C",
773 "[": "C",
774 "]": "C",
775 " ": "C",
776 "\\": "C",
777 "/": "C",
778 "?": "C",
779 }
780 )
782 def bindparam_string(self, name, **kw):
783 quote = getattr(name, "quote", None)
784 if (
785 quote is True
786 or quote is not False
787 and self.preparer._bindparam_requires_quotes(name)
788 # bind param quoting for Oracle doesn't work with post_compile
789 # params. For those, the default bindparam_string will escape
790 # special chars, and the appending of a number "_1" etc. will
791 # take care of reserved words
792 and not kw.get("post_compile", False)
793 ):
794 # interesting to note about expanding parameters - since the
795 # new parameters take the form <paramname>_<int>, at least if
796 # they are originally formed from reserved words, they no longer
797 # need quoting :). names that include illegal characters
798 # won't work however.
799 quoted_name = '"%s"' % name
800 kw["escaped_from"] = name
801 name = quoted_name
802 return OracleCompiler.bindparam_string(self, name, **kw)
804 # TODO: we could likely do away with quoting altogether for
805 # Oracle parameters and use the custom escaping here
806 escaped_from = kw.get("escaped_from", None)
807 if not escaped_from:
808 if self._bind_translate_re.search(name):
809 # not quite the translate use case as we want to
810 # also get a quick boolean if we even found
811 # unusual characters in the name
812 new_name = self._bind_translate_re.sub(
813 lambda m: self._bind_translate_chars[m.group(0)],
814 name,
815 )
816 if new_name[0].isdigit() or new_name[0] == "_":
817 new_name = "D" + new_name
818 kw["escaped_from"] = name
819 name = new_name
820 elif name[0].isdigit() or name[0] == "_":
821 new_name = "D" + name
822 kw["escaped_from"] = name
823 name = new_name
825 return OracleCompiler.bindparam_string(self, name, **kw)
828class OracleExecutionContext_cx_oracle(OracleExecutionContext):
829 out_parameters = None
831 def _generate_out_parameter_vars(self):
832 # check for has_out_parameters or RETURNING, create cx_Oracle.var
833 # objects if so
834 if self.compiled.has_out_parameters or self.compiled._oracle_returning:
835 out_parameters = self.out_parameters
836 assert out_parameters is not None
838 len_params = len(self.parameters)
840 quoted_bind_names = self.compiled.escaped_bind_names
841 for bindparam in self.compiled.binds.values():
842 if bindparam.isoutparam:
843 name = self.compiled.bind_names[bindparam]
844 type_impl = bindparam.type.dialect_impl(self.dialect)
846 if hasattr(type_impl, "_cx_oracle_var"):
847 out_parameters[name] = type_impl._cx_oracle_var(
848 self.dialect, self.cursor, arraysize=len_params
849 )
850 else:
851 dbtype = type_impl.get_dbapi_type(self.dialect.dbapi)
853 cx_Oracle = self.dialect.dbapi
855 assert cx_Oracle is not None
857 if dbtype is None:
858 raise exc.InvalidRequestError(
859 "Cannot create out parameter for "
860 "parameter "
861 "%r - its type %r is not supported by"
862 " cx_oracle" % (bindparam.key, bindparam.type)
863 )
865 # note this is an OUT parameter. Using
866 # non-LOB datavalues with large unicode-holding
867 # values causes the failure (both cx_Oracle and
868 # oracledb):
869 # ORA-22835: Buffer too small for CLOB to CHAR or
870 # BLOB to RAW conversion (actual: 16507,
871 # maximum: 4000)
872 # [SQL: INSERT INTO long_text (x, y, z) VALUES
873 # (:x, :y, :z) RETURNING long_text.x, long_text.y,
874 # long_text.z INTO :ret_0, :ret_1, :ret_2]
875 # so even for DB_TYPE_NVARCHAR we convert to a LOB
877 if isinstance(type_impl, _LOBDataType):
878 if dbtype == cx_Oracle.DB_TYPE_NVARCHAR:
879 dbtype = cx_Oracle.NCLOB
880 elif dbtype == cx_Oracle.DB_TYPE_RAW:
881 dbtype = cx_Oracle.BLOB
882 # other LOB types go in directly
884 out_parameters[name] = self.cursor.var(
885 dbtype,
886 # this is fine also in oracledb_async since
887 # the driver will await the read coroutine
888 outconverter=lambda value: value.read(),
889 arraysize=len_params,
890 )
891 elif (
892 isinstance(type_impl, _OracleNumeric)
893 and type_impl.asdecimal
894 ):
895 out_parameters[name] = self.cursor.var(
896 decimal.Decimal,
897 arraysize=len_params,
898 )
900 else:
901 out_parameters[name] = self.cursor.var(
902 dbtype, arraysize=len_params
903 )
905 for param in self.parameters:
906 param[quoted_bind_names.get(name, name)] = (
907 out_parameters[name]
908 )
910 def _generate_cursor_outputtype_handler(self):
911 output_handlers = {}
913 for keyname, name, objects, type_ in self.compiled._result_columns:
914 handler = type_._cached_custom_processor(
915 self.dialect,
916 "cx_oracle_outputtypehandler",
917 self._get_cx_oracle_type_handler,
918 )
920 if handler:
921 denormalized_name = self.dialect.denormalize_name(keyname)
922 output_handlers[denormalized_name] = handler
924 if output_handlers:
925 default_handler = self._dbapi_connection.outputtypehandler
927 def output_type_handler(
928 cursor, name, default_type, size, precision, scale
929 ):
930 if name in output_handlers:
931 return output_handlers[name](
932 cursor, name, default_type, size, precision, scale
933 )
934 else:
935 return default_handler(
936 cursor, name, default_type, size, precision, scale
937 )
939 self.cursor.outputtypehandler = output_type_handler
941 def _get_cx_oracle_type_handler(self, impl):
942 if hasattr(impl, "_cx_oracle_outputtypehandler"):
943 return impl._cx_oracle_outputtypehandler(self.dialect)
944 else:
945 return None
947 def pre_exec(self):
948 super().pre_exec()
949 if not getattr(self.compiled, "_oracle_cx_sql_compiler", False):
950 return
952 self.out_parameters = {}
954 self._generate_out_parameter_vars()
956 self._generate_cursor_outputtype_handler()
958 def post_exec(self):
959 if (
960 self.compiled
961 and is_sql_compiler(self.compiled)
962 and self.compiled._oracle_returning
963 ):
964 initial_buffer = self.fetchall_for_returning(
965 self.cursor, _internal=True
966 )
968 fetch_strategy = _cursor.FullyBufferedCursorFetchStrategy(
969 self.cursor,
970 [
971 (entry.keyname, None)
972 for entry in self.compiled._result_columns
973 ],
974 initial_buffer=initial_buffer,
975 )
977 self.cursor_fetch_strategy = fetch_strategy
979 def create_cursor(self):
980 c = self._dbapi_connection.cursor()
981 if self.dialect.arraysize:
982 c.arraysize = self.dialect.arraysize
984 return c
986 def fetchall_for_returning(self, cursor, *, _internal=False):
987 compiled = self.compiled
988 if (
989 not _internal
990 and compiled is None
991 or not is_sql_compiler(compiled)
992 or not compiled._oracle_returning
993 ):
994 raise NotImplementedError(
995 "execution context was not prepared for Oracle RETURNING"
996 )
998 # create a fake cursor result from the out parameters. unlike
999 # get_out_parameter_values(), the result-row handlers here will be
1000 # applied at the Result level
1002 numcols = len(self.out_parameters)
1004 # [stmt_result for stmt_result in outparam.values] == each
1005 # statement in executemany
1006 # [val for val in stmt_result] == each row for a particular
1007 # statement
1008 return list(
1009 zip(
1010 *[
1011 [
1012 val
1013 for stmt_result in self.out_parameters[
1014 f"ret_{j}"
1015 ].values
1016 for val in (stmt_result or ())
1017 ]
1018 for j in range(numcols)
1019 ]
1020 )
1021 )
1023 def get_out_parameter_values(self, out_param_names):
1024 # this method should not be called when the compiler has
1025 # RETURNING as we've turned the has_out_parameters flag set to
1026 # False.
1027 assert not self.compiled.returning
1029 return [
1030 self.dialect._paramval(self.out_parameters[name])
1031 for name in out_param_names
1032 ]
1035class OracleDialect_cx_oracle(OracleDialect):
1036 supports_statement_cache = True
1037 execution_ctx_cls = OracleExecutionContext_cx_oracle
1038 statement_compiler = OracleCompiler_cx_oracle
1040 supports_sane_rowcount = True
1041 supports_sane_multi_rowcount = True
1043 insert_executemany_returning = True
1044 insert_executemany_returning_sort_by_parameter_order = True
1045 update_executemany_returning = True
1046 delete_executemany_returning = True
1048 bind_typing = interfaces.BindTyping.SETINPUTSIZES
1050 driver = "cx_oracle"
1052 colspecs = util.update_copy(
1053 OracleDialect.colspecs,
1054 {
1055 sqltypes.TIMESTAMP: _CXOracleTIMESTAMP,
1056 sqltypes.Numeric: _OracleNumeric,
1057 sqltypes.Float: _OracleNumeric,
1058 oracle.BINARY_FLOAT: _OracleBINARY_FLOAT,
1059 oracle.BINARY_DOUBLE: _OracleBINARY_DOUBLE,
1060 sqltypes.Integer: _OracleInteger,
1061 oracle.NUMBER: _OracleNUMBER,
1062 sqltypes.Date: _CXOracleDate,
1063 sqltypes.LargeBinary: _OracleBinary,
1064 sqltypes.Boolean: oracle._OracleBoolean,
1065 sqltypes.Interval: _OracleInterval,
1066 oracle.INTERVAL: _OracleInterval,
1067 sqltypes.Text: _OracleText,
1068 sqltypes.String: _OracleString,
1069 sqltypes.UnicodeText: _OracleUnicodeTextCLOB,
1070 sqltypes.CHAR: _OracleChar,
1071 sqltypes.NCHAR: _OracleNChar,
1072 sqltypes.Enum: _OracleEnum,
1073 oracle.LONG: _OracleLong,
1074 oracle.RAW: _OracleRaw,
1075 sqltypes.Unicode: _OracleUnicodeStringCHAR,
1076 sqltypes.NVARCHAR: _OracleUnicodeStringNCHAR,
1077 sqltypes.Uuid: _OracleUUID,
1078 oracle.NCLOB: _OracleUnicodeTextNCLOB,
1079 oracle.ROWID: _OracleRowid,
1080 },
1081 )
1083 execute_sequence_format = list
1085 _cx_oracle_threaded = None
1087 _cursor_var_unicode_kwargs = util.immutabledict()
1089 @util.deprecated_params(
1090 threaded=(
1091 "1.3",
1092 "The 'threaded' parameter to the cx_oracle/oracledb dialect "
1093 "is deprecated as a dialect-level argument, and will be removed "
1094 "in a future release. As of version 1.3, it defaults to False "
1095 "rather than True. The 'threaded' option can be passed to "
1096 "cx_Oracle directly in the URL query string passed to "
1097 ":func:`_sa.create_engine`.",
1098 )
1099 )
1100 def __init__(
1101 self,
1102 auto_convert_lobs=True,
1103 coerce_to_decimal=True,
1104 arraysize=None,
1105 encoding_errors=None,
1106 threaded=None,
1107 **kwargs,
1108 ):
1109 OracleDialect.__init__(self, **kwargs)
1110 self.arraysize = arraysize
1111 self.encoding_errors = encoding_errors
1112 if encoding_errors:
1113 self._cursor_var_unicode_kwargs = {
1114 "encodingErrors": encoding_errors
1115 }
1116 if threaded is not None:
1117 self._cx_oracle_threaded = threaded
1118 self.auto_convert_lobs = auto_convert_lobs
1119 self.coerce_to_decimal = coerce_to_decimal
1120 if self._use_nchar_for_unicode:
1121 self.colspecs = self.colspecs.copy()
1122 self.colspecs[sqltypes.Unicode] = _OracleUnicodeStringNCHAR
1123 self.colspecs[sqltypes.UnicodeText] = _OracleUnicodeTextNCLOB
1125 dbapi_module = self.dbapi
1126 self._load_version(dbapi_module)
1128 if dbapi_module is not None:
1129 # these constants will first be seen in SQLAlchemy datatypes
1130 # coming from the get_dbapi_type() method. We then
1131 # will place the following types into setinputsizes() calls
1132 # on each statement. Oracle constants that are not in this
1133 # list will not be put into setinputsizes().
1134 self.include_set_input_sizes = {
1135 dbapi_module.DATETIME,
1136 dbapi_module.DB_TYPE_NVARCHAR, # used for CLOB, NCLOB
1137 dbapi_module.DB_TYPE_RAW, # used for BLOB
1138 dbapi_module.NCLOB, # not currently used except for OUT param
1139 dbapi_module.CLOB, # not currently used except for OUT param
1140 dbapi_module.LOB, # not currently used
1141 dbapi_module.BLOB, # not currently used except for OUT param
1142 dbapi_module.NCHAR,
1143 dbapi_module.FIXED_NCHAR,
1144 dbapi_module.FIXED_CHAR,
1145 dbapi_module.TIMESTAMP,
1146 int, # _OracleInteger,
1147 # _OracleBINARY_FLOAT, _OracleBINARY_DOUBLE,
1148 dbapi_module.NATIVE_FLOAT,
1149 }
1151 self._paramval = lambda value: value.getvalue()
1153 def _load_version(self, dbapi_module):
1154 version = (0, 0, 0)
1155 if dbapi_module is not None:
1156 m = re.match(r"(\d+)\.(\d+)(?:\.(\d+))?", dbapi_module.version)
1157 if m:
1158 version = tuple(
1159 int(x) for x in m.group(1, 2, 3) if x is not None
1160 )
1161 self.cx_oracle_ver = version
1162 if self.cx_oracle_ver < (8,) and self.cx_oracle_ver > (0, 0, 0):
1163 raise exc.InvalidRequestError(
1164 "cx_Oracle version 8 and above are supported"
1165 )
1167 @classmethod
1168 def import_dbapi(cls):
1169 import cx_Oracle
1171 return cx_Oracle
1173 def initialize(self, connection):
1174 super().initialize(connection)
1175 self._detect_decimal_char(connection)
1177 def get_isolation_level(self, dbapi_connection):
1178 # sources:
1180 # general idea of transaction id, have to start one, etc.
1181 # https://stackoverflow.com/questions/10711204/how-to-check-isoloation-level
1183 # how to decode xid cols from v$transaction to match
1184 # https://asktom.oracle.com/pls/apex/f?p=100:11:0::::P11_QUESTION_ID:9532779900346079444
1186 # Oracle tuple comparison without using IN:
1187 # https://www.sql-workbench.eu/comparison/tuple_comparison.html
1189 with dbapi_connection.cursor() as cursor:
1190 # this is the only way to ensure a transaction is started without
1191 # actually running DML. There's no way to see the configured
1192 # isolation level without getting it from v$transaction which
1193 # means transaction has to be started.
1194 outval = cursor.var(str)
1195 cursor.execute(
1196 """
1197 begin
1198 :trans_id := dbms_transaction.local_transaction_id( TRUE );
1199 end;
1200 """,
1201 {"trans_id": outval},
1202 )
1203 trans_id = outval.getvalue()
1204 xidusn, xidslot, xidsqn = trans_id.split(".", 2)
1206 cursor.execute(
1207 "SELECT CASE BITAND(t.flag, POWER(2, 28)) "
1208 "WHEN 0 THEN 'READ COMMITTED' "
1209 "ELSE 'SERIALIZABLE' END AS isolation_level "
1210 "FROM v$transaction t WHERE "
1211 "(t.xidusn, t.xidslot, t.xidsqn) = "
1212 "((:xidusn, :xidslot, :xidsqn))",
1213 {"xidusn": xidusn, "xidslot": xidslot, "xidsqn": xidsqn},
1214 )
1215 row = cursor.fetchone()
1216 if row is None:
1217 raise exc.InvalidRequestError(
1218 "could not retrieve isolation level"
1219 )
1220 result = row[0]
1222 return result
1224 def get_isolation_level_values(self, dbapi_connection):
1225 return super().get_isolation_level_values(dbapi_connection) + [
1226 "AUTOCOMMIT"
1227 ]
1229 def set_isolation_level(self, dbapi_connection, level):
1230 if level == "AUTOCOMMIT":
1231 dbapi_connection.autocommit = True
1232 else:
1233 dbapi_connection.autocommit = False
1234 dbapi_connection.rollback()
1235 with dbapi_connection.cursor() as cursor:
1236 cursor.execute(f"ALTER SESSION SET ISOLATION_LEVEL={level}")
1238 def detect_autocommit_setting(self, dbapi_conn) -> bool:
1239 return bool(dbapi_conn.autocommit)
1241 def _detect_decimal_char(self, connection):
1242 # we have the option to change this setting upon connect,
1243 # or just look at what it is upon connect and convert.
1244 # to minimize the chance of interference with changes to
1245 # NLS_TERRITORY or formatting behavior of the DB, we opt
1246 # to just look at it
1248 dbapi_connection = connection.connection
1250 with dbapi_connection.cursor() as cursor:
1251 # issue #8744
1252 # nls_session_parameters is not available in some Oracle
1253 # modes like "mount mode". But then, v$nls_parameters is not
1254 # available if the connection doesn't have SYSDBA priv.
1255 #
1256 # simplify the whole thing and just use the method that we were
1257 # doing in the test suite already, selecting a number
1259 def output_type_handler(
1260 cursor, name, defaultType, size, precision, scale
1261 ):
1262 return cursor.var(
1263 self.dbapi.STRING, 255, arraysize=cursor.arraysize
1264 )
1266 cursor.outputtypehandler = output_type_handler
1267 cursor.execute("SELECT 1.1 FROM DUAL")
1268 value = cursor.fetchone()[0]
1270 decimal_char = value.lstrip("0")[1]
1271 assert not decimal_char[0].isdigit()
1273 self._decimal_char = decimal_char
1275 if self._decimal_char != ".":
1276 _detect_decimal = self._detect_decimal
1277 _to_decimal = self._to_decimal
1279 self._detect_decimal = lambda value: _detect_decimal(
1280 value.replace(self._decimal_char, ".")
1281 )
1282 self._to_decimal = lambda value: _to_decimal(
1283 value.replace(self._decimal_char, ".")
1284 )
1286 def _detect_decimal(self, value):
1287 if "." in value:
1288 return self._to_decimal(value)
1289 else:
1290 return int(value)
1292 _to_decimal = decimal.Decimal
1294 def _generate_connection_outputtype_handler(self):
1295 """establish the default outputtypehandler established at the
1296 connection level.
1298 """
1300 dialect = self
1301 cx_Oracle = dialect.dbapi
1303 number_handler = _OracleNUMBER(
1304 asdecimal=True
1305 )._cx_oracle_outputtypehandler(dialect)
1306 float_handler = _OracleNUMBER(
1307 asdecimal=False
1308 )._cx_oracle_outputtypehandler(dialect)
1310 def output_type_handler(
1311 cursor, name, default_type, size, precision, scale
1312 ):
1313 if (
1314 default_type == cx_Oracle.NUMBER
1315 and default_type is not cx_Oracle.NATIVE_FLOAT
1316 ):
1317 if not dialect.coerce_to_decimal:
1318 return None
1319 elif precision == 0 and scale in (0, -127):
1320 # ambiguous type, this occurs when selecting
1321 # numbers from deep subqueries
1322 return cursor.var(
1323 cx_Oracle.STRING,
1324 255,
1325 outconverter=dialect._detect_decimal,
1326 arraysize=cursor.arraysize,
1327 )
1328 elif precision and scale > 0:
1329 return number_handler(
1330 cursor, name, default_type, size, precision, scale
1331 )
1332 else:
1333 return float_handler(
1334 cursor, name, default_type, size, precision, scale
1335 )
1337 # if unicode options were specified, add a decoder, otherwise
1338 # cx_Oracle should return Unicode
1339 elif (
1340 dialect._cursor_var_unicode_kwargs
1341 and default_type
1342 in (
1343 cx_Oracle.STRING,
1344 cx_Oracle.FIXED_CHAR,
1345 )
1346 and default_type is not cx_Oracle.CLOB
1347 and default_type is not cx_Oracle.NCLOB
1348 ):
1349 return cursor.var(
1350 str,
1351 size,
1352 cursor.arraysize,
1353 **dialect._cursor_var_unicode_kwargs,
1354 )
1356 elif dialect.auto_convert_lobs and default_type in (
1357 cx_Oracle.CLOB,
1358 cx_Oracle.NCLOB,
1359 ):
1360 typ = (
1361 cx_Oracle.DB_TYPE_VARCHAR
1362 if default_type is cx_Oracle.CLOB
1363 else cx_Oracle.DB_TYPE_NVARCHAR
1364 )
1365 return cursor.var(
1366 typ,
1367 _CX_ORACLE_MAGIC_LOB_SIZE,
1368 cursor.arraysize,
1369 **dialect._cursor_var_unicode_kwargs,
1370 )
1372 elif dialect.auto_convert_lobs and default_type in (
1373 cx_Oracle.BLOB,
1374 ):
1375 return cursor.var(
1376 cx_Oracle.DB_TYPE_RAW,
1377 _CX_ORACLE_MAGIC_LOB_SIZE,
1378 cursor.arraysize,
1379 )
1381 return output_type_handler
1383 def on_connect(self):
1384 output_type_handler = self._generate_connection_outputtype_handler()
1386 def on_connect(conn):
1387 conn.outputtypehandler = output_type_handler
1389 return on_connect
1391 def create_connect_args(self, url):
1392 opts = dict(url.query)
1394 for opt in ("use_ansi", "auto_convert_lobs"):
1395 if opt in opts:
1396 util.warn_deprecated(
1397 f"{self.driver} dialect option {opt!r} should only be "
1398 "passed to create_engine directly, not within the URL "
1399 "string",
1400 version="1.3",
1401 )
1402 util.coerce_kw_type(opts, opt, bool)
1403 setattr(self, opt, opts.pop(opt))
1405 database = url.database
1406 service_name = opts.pop("service_name", None)
1407 if database or service_name:
1408 # if we have a database, then we have a remote host
1409 port = url.port
1410 if port:
1411 port = int(port)
1412 else:
1413 port = 1521
1415 if database and service_name:
1416 raise exc.InvalidRequestError(
1417 '"service_name" option shouldn\'t '
1418 'be used with a "database" part of the url'
1419 )
1420 if database:
1421 makedsn_kwargs = {"sid": database}
1422 if service_name:
1423 makedsn_kwargs = {"service_name": service_name}
1425 dsn = self.dbapi.makedsn(url.host, port, **makedsn_kwargs)
1426 else:
1427 # we have a local tnsname
1428 dsn = url.host
1430 if dsn is not None:
1431 opts["dsn"] = dsn
1432 if url.password is not None:
1433 opts["password"] = url.password
1434 if url.username is not None:
1435 opts["user"] = url.username
1437 if self._cx_oracle_threaded is not None:
1438 opts.setdefault("threaded", self._cx_oracle_threaded)
1440 def convert_cx_oracle_constant(value):
1441 if isinstance(value, str):
1442 try:
1443 int_val = int(value)
1444 except ValueError:
1445 value = value.upper()
1446 return getattr(self.dbapi, value)
1447 else:
1448 return int_val
1449 else:
1450 return value
1452 util.coerce_kw_type(opts, "mode", convert_cx_oracle_constant)
1453 util.coerce_kw_type(opts, "threaded", bool)
1454 util.coerce_kw_type(opts, "events", bool)
1455 util.coerce_kw_type(opts, "purity", convert_cx_oracle_constant)
1456 return ([], opts)
1458 def _get_server_version_info(self, connection):
1459 return tuple(int(x) for x in connection.connection.version.split("."))
1461 def is_disconnect(self, e, connection, cursor):
1462 (error,) = e.args
1463 if isinstance(
1464 e, (self.dbapi.InterfaceError, self.dbapi.DatabaseError)
1465 ) and "not connected" in str(e):
1466 return True
1468 if hasattr(error, "code") and error.code in {
1469 28,
1470 3114,
1471 3113,
1472 3135,
1473 1033,
1474 2396,
1475 }:
1476 # ORA-00028: your session has been killed
1477 # ORA-03114: not connected to ORACLE
1478 # ORA-03113: end-of-file on communication channel
1479 # ORA-03135: connection lost contact
1480 # ORA-01033: ORACLE initialization or shutdown in progress
1481 # ORA-02396: exceeded maximum idle time, please connect again
1482 # TODO: Others ?
1483 return True
1485 if re.match(r"^(?:DPI-1010|DPI-1080|DPY-1001|DPY-4011)", str(e)):
1486 # DPI-1010: not connected
1487 # DPI-1080: connection was closed by ORA-3113
1488 # python-oracledb's DPY-1001: not connected to database
1489 # python-oracledb's DPY-4011: the database or network closed the
1490 # connection
1491 # TODO: others?
1492 return True
1494 return False
1496 def create_xid(self):
1497 id_ = random.randint(0, 2**128)
1498 return (0x1234, "%032x" % id_, "%032x" % 9)
1500 def do_executemany(self, cursor, statement, parameters, context=None):
1501 if isinstance(parameters, tuple):
1502 parameters = list(parameters)
1503 cursor.executemany(statement, parameters)
1505 def do_begin_twophase(self, connection, xid):
1506 connection.connection.begin(*xid)
1507 connection.connection.info["cx_oracle_xid"] = xid
1509 def do_prepare_twophase(self, connection, xid):
1510 result = connection.connection.prepare()
1511 connection.info["cx_oracle_prepared"] = result
1513 def do_rollback_twophase(
1514 self, connection, xid, is_prepared=True, recover=False
1515 ):
1516 self.do_rollback(connection.connection)
1517 # TODO: need to end XA state here
1519 def do_commit_twophase(
1520 self, connection, xid, is_prepared=True, recover=False
1521 ):
1522 if not is_prepared:
1523 self.do_commit(connection.connection)
1524 else:
1525 if recover:
1526 raise NotImplementedError(
1527 "2pc recovery not implemented for cx_Oracle"
1528 )
1529 oci_prepared = connection.info["cx_oracle_prepared"]
1530 if oci_prepared:
1531 self.do_commit(connection.connection)
1532 # TODO: need to end XA state here
1534 def do_set_input_sizes(self, cursor, list_of_tuples, context):
1535 if self.positional:
1536 # not usually used, here to support if someone is modifying
1537 # the dialect to use positional style
1538 cursor.setinputsizes(
1539 *[dbtype for key, dbtype, sqltype in list_of_tuples]
1540 )
1541 else:
1542 collection = (
1543 (key, dbtype)
1544 for key, dbtype, sqltype in list_of_tuples
1545 if dbtype
1546 )
1548 cursor.setinputsizes(**{key: dbtype for key, dbtype in collection})
1550 def do_recover_twophase(self, connection):
1551 raise NotImplementedError(
1552 "recover two phase query for cx_Oracle not implemented"
1553 )
1556dialect = OracleDialect_cx_oracle