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-2025 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
492from __future__ import annotations
494import decimal
495import random
496import re
498from . import base as oracle
499from .base import OracleCompiler
500from .base import OracleDialect
501from .base import OracleExecutionContext
502from .types import _OracleDateLiteralRender
503from ... import exc
504from ... import util
505from ...engine import cursor as _cursor
506from ...engine import interfaces
507from ...engine import processors
508from ...sql import sqltypes
509from ...sql._typing import is_sql_compiler
511# source:
512# https://github.com/oracle/python-cx_Oracle/issues/596#issuecomment-999243649
513_CX_ORACLE_MAGIC_LOB_SIZE = 131072
516class _OracleInteger(sqltypes.Integer):
517 def get_dbapi_type(self, dbapi):
518 # see https://github.com/oracle/python-cx_Oracle/issues/
519 # 208#issuecomment-409715955
520 return int
522 def _cx_oracle_var(self, dialect, cursor, arraysize=None):
523 cx_Oracle = dialect.dbapi
524 return cursor.var(
525 cx_Oracle.STRING,
526 255,
527 arraysize=arraysize if arraysize is not None else cursor.arraysize,
528 outconverter=int,
529 )
531 def _cx_oracle_outputtypehandler(self, dialect):
532 def handler(cursor, name, default_type, size, precision, scale):
533 return self._cx_oracle_var(dialect, cursor)
535 return handler
538class _OracleNumeric(sqltypes.Numeric):
539 is_number = False
541 def bind_processor(self, dialect):
542 if self.scale == 0:
543 return None
544 elif self.asdecimal:
545 processor = processors.to_decimal_processor_factory(
546 decimal.Decimal, self._effective_decimal_return_scale
547 )
549 def process(value):
550 if isinstance(value, (int, float)):
551 return processor(value)
552 elif value is not None and value.is_infinite():
553 return float(value)
554 else:
555 return value
557 return process
558 else:
559 return processors.to_float
561 def result_processor(self, dialect, coltype):
562 return None
564 def _cx_oracle_outputtypehandler(self, dialect):
565 cx_Oracle = dialect.dbapi
567 def handler(cursor, name, default_type, size, precision, scale):
568 outconverter = None
570 if precision:
571 if self.asdecimal:
572 if default_type == cx_Oracle.NATIVE_FLOAT:
573 # receiving float and doing Decimal after the fact
574 # allows for float("inf") to be handled
575 type_ = default_type
576 outconverter = decimal.Decimal
577 else:
578 type_ = decimal.Decimal
579 else:
580 if self.is_number and scale == 0:
581 # integer. cx_Oracle is observed to handle the widest
582 # variety of ints when no directives are passed,
583 # from 5.2 to 7.0. See [ticket:4457]
584 return None
585 else:
586 type_ = cx_Oracle.NATIVE_FLOAT
588 else:
589 if self.asdecimal:
590 if default_type == cx_Oracle.NATIVE_FLOAT:
591 type_ = default_type
592 outconverter = decimal.Decimal
593 else:
594 type_ = decimal.Decimal
595 else:
596 if self.is_number and scale == 0:
597 # integer. cx_Oracle is observed to handle the widest
598 # variety of ints when no directives are passed,
599 # from 5.2 to 7.0. See [ticket:4457]
600 return None
601 else:
602 type_ = cx_Oracle.NATIVE_FLOAT
604 return cursor.var(
605 type_,
606 255,
607 arraysize=cursor.arraysize,
608 outconverter=outconverter,
609 )
611 return handler
614class _OracleUUID(sqltypes.Uuid):
615 def get_dbapi_type(self, dbapi):
616 return dbapi.STRING
619class _OracleBinaryFloat(_OracleNumeric):
620 def get_dbapi_type(self, dbapi):
621 return dbapi.NATIVE_FLOAT
624class _OracleBINARY_FLOAT(_OracleBinaryFloat, oracle.BINARY_FLOAT):
625 pass
628class _OracleBINARY_DOUBLE(_OracleBinaryFloat, oracle.BINARY_DOUBLE):
629 pass
632class _OracleNUMBER(_OracleNumeric):
633 is_number = True
636class _CXOracleDate(oracle._OracleDate):
637 def bind_processor(self, dialect):
638 return None
640 def result_processor(self, dialect, coltype):
641 def process(value):
642 if value is not None:
643 return value.date()
644 else:
645 return value
647 return process
650class _CXOracleTIMESTAMP(_OracleDateLiteralRender, sqltypes.TIMESTAMP):
651 def literal_processor(self, dialect):
652 return self._literal_processor_datetime(dialect)
655class _LOBDataType:
656 pass
659# TODO: the names used across CHAR / VARCHAR / NCHAR / NVARCHAR
660# here are inconsistent and not very good
661class _OracleChar(sqltypes.CHAR):
662 def get_dbapi_type(self, dbapi):
663 return dbapi.FIXED_CHAR
666class _OracleNChar(sqltypes.NCHAR):
667 def get_dbapi_type(self, dbapi):
668 return dbapi.FIXED_NCHAR
671class _OracleUnicodeStringNCHAR(oracle.NVARCHAR2):
672 def get_dbapi_type(self, dbapi):
673 return dbapi.NCHAR
676class _OracleUnicodeStringCHAR(sqltypes.Unicode):
677 def get_dbapi_type(self, dbapi):
678 return dbapi.LONG_STRING
681class _OracleUnicodeTextNCLOB(_LOBDataType, oracle.NCLOB):
682 def get_dbapi_type(self, dbapi):
683 # previously, this was dbapi.NCLOB.
684 # DB_TYPE_NVARCHAR will instead be passed to setinputsizes()
685 # when this datatype is used.
686 return dbapi.DB_TYPE_NVARCHAR
689class _OracleUnicodeTextCLOB(_LOBDataType, sqltypes.UnicodeText):
690 def get_dbapi_type(self, dbapi):
691 # previously, this was dbapi.CLOB.
692 # DB_TYPE_NVARCHAR will instead be passed to setinputsizes()
693 # when this datatype is used.
694 return dbapi.DB_TYPE_NVARCHAR
697class _OracleText(_LOBDataType, sqltypes.Text):
698 def get_dbapi_type(self, dbapi):
699 # previously, this was dbapi.CLOB.
700 # DB_TYPE_NVARCHAR will instead be passed to setinputsizes()
701 # when this datatype is used.
702 return dbapi.DB_TYPE_NVARCHAR
705class _OracleLong(_LOBDataType, oracle.LONG):
706 def get_dbapi_type(self, dbapi):
707 return dbapi.LONG_STRING
710class _OracleString(sqltypes.String):
711 pass
714class _OracleEnum(sqltypes.Enum):
715 def bind_processor(self, dialect):
716 enum_proc = sqltypes.Enum.bind_processor(self, dialect)
718 def process(value):
719 raw_str = enum_proc(value)
720 return raw_str
722 return process
725class _OracleBinary(_LOBDataType, sqltypes.LargeBinary):
726 def get_dbapi_type(self, dbapi):
727 # previously, this was dbapi.BLOB.
728 # DB_TYPE_RAW will instead be passed to setinputsizes()
729 # when this datatype is used.
730 return dbapi.DB_TYPE_RAW
732 def bind_processor(self, dialect):
733 return None
735 def result_processor(self, dialect, coltype):
736 if not dialect.auto_convert_lobs:
737 return None
738 else:
739 return super().result_processor(dialect, coltype)
742class _OracleInterval(oracle.INTERVAL):
743 def get_dbapi_type(self, dbapi):
744 return dbapi.INTERVAL
747class _OracleRaw(oracle.RAW):
748 pass
751class _OracleRowid(oracle.ROWID):
752 def get_dbapi_type(self, dbapi):
753 return dbapi.ROWID
756class OracleCompiler_cx_oracle(OracleCompiler):
757 _oracle_cx_sql_compiler = True
759 _oracle_returning = False
761 # Oracle bind names can't start with digits or underscores.
762 # currently we rely upon Oracle-specific quoting of bind names in most
763 # cases. however for expanding params, the escape chars are used.
764 # see #8708
765 bindname_escape_characters = util.immutabledict(
766 {
767 "%": "P",
768 "(": "A",
769 ")": "Z",
770 ":": "C",
771 ".": "C",
772 "[": "C",
773 "]": "C",
774 " ": "C",
775 "\\": "C",
776 "/": "C",
777 "?": "C",
778 }
779 )
781 def bindparam_string(self, name, **kw):
782 quote = getattr(name, "quote", None)
783 if (
784 quote is True
785 or quote is not False
786 and self.preparer._bindparam_requires_quotes(name)
787 # bind param quoting for Oracle doesn't work with post_compile
788 # params. For those, the default bindparam_string will escape
789 # special chars, and the appending of a number "_1" etc. will
790 # take care of reserved words
791 and not kw.get("post_compile", False)
792 ):
793 # interesting to note about expanding parameters - since the
794 # new parameters take the form <paramname>_<int>, at least if
795 # they are originally formed from reserved words, they no longer
796 # need quoting :). names that include illegal characters
797 # won't work however.
798 quoted_name = '"%s"' % name
799 kw["escaped_from"] = name
800 name = quoted_name
801 return OracleCompiler.bindparam_string(self, name, **kw)
803 # TODO: we could likely do away with quoting altogether for
804 # Oracle parameters and use the custom escaping here
805 escaped_from = kw.get("escaped_from", None)
806 if not escaped_from:
807 if self._bind_translate_re.search(name):
808 # not quite the translate use case as we want to
809 # also get a quick boolean if we even found
810 # unusual characters in the name
811 new_name = self._bind_translate_re.sub(
812 lambda m: self._bind_translate_chars[m.group(0)],
813 name,
814 )
815 if new_name[0].isdigit() or new_name[0] == "_":
816 new_name = "D" + new_name
817 kw["escaped_from"] = name
818 name = new_name
819 elif name[0].isdigit() or name[0] == "_":
820 new_name = "D" + name
821 kw["escaped_from"] = name
822 name = new_name
824 return OracleCompiler.bindparam_string(self, name, **kw)
827class OracleExecutionContext_cx_oracle(OracleExecutionContext):
828 out_parameters = None
830 def _generate_out_parameter_vars(self):
831 # check for has_out_parameters or RETURNING, create cx_Oracle.var
832 # objects if so
833 if self.compiled.has_out_parameters or self.compiled._oracle_returning:
834 out_parameters = self.out_parameters
835 assert out_parameters is not None
837 len_params = len(self.parameters)
839 quoted_bind_names = self.compiled.escaped_bind_names
840 for bindparam in self.compiled.binds.values():
841 if bindparam.isoutparam:
842 name = self.compiled.bind_names[bindparam]
843 type_impl = bindparam.type.dialect_impl(self.dialect)
845 if hasattr(type_impl, "_cx_oracle_var"):
846 out_parameters[name] = type_impl._cx_oracle_var(
847 self.dialect, self.cursor, arraysize=len_params
848 )
849 else:
850 dbtype = type_impl.get_dbapi_type(self.dialect.dbapi)
852 cx_Oracle = self.dialect.dbapi
854 assert cx_Oracle is not None
856 if dbtype is None:
857 raise exc.InvalidRequestError(
858 "Cannot create out parameter for "
859 "parameter "
860 "%r - its type %r is not supported by"
861 " cx_oracle" % (bindparam.key, bindparam.type)
862 )
864 # note this is an OUT parameter. Using
865 # non-LOB datavalues with large unicode-holding
866 # values causes the failure (both cx_Oracle and
867 # oracledb):
868 # ORA-22835: Buffer too small for CLOB to CHAR or
869 # BLOB to RAW conversion (actual: 16507,
870 # maximum: 4000)
871 # [SQL: INSERT INTO long_text (x, y, z) VALUES
872 # (:x, :y, :z) RETURNING long_text.x, long_text.y,
873 # long_text.z INTO :ret_0, :ret_1, :ret_2]
874 # so even for DB_TYPE_NVARCHAR we convert to a LOB
876 if isinstance(type_impl, _LOBDataType):
877 if dbtype == cx_Oracle.DB_TYPE_NVARCHAR:
878 dbtype = cx_Oracle.NCLOB
879 elif dbtype == cx_Oracle.DB_TYPE_RAW:
880 dbtype = cx_Oracle.BLOB
881 # other LOB types go in directly
883 out_parameters[name] = self.cursor.var(
884 dbtype,
885 # this is fine also in oracledb_async since
886 # the driver will await the read coroutine
887 outconverter=lambda value: value.read(),
888 arraysize=len_params,
889 )
890 elif (
891 isinstance(type_impl, _OracleNumeric)
892 and type_impl.asdecimal
893 ):
894 out_parameters[name] = self.cursor.var(
895 decimal.Decimal,
896 arraysize=len_params,
897 )
899 else:
900 out_parameters[name] = self.cursor.var(
901 dbtype, arraysize=len_params
902 )
904 for param in self.parameters:
905 param[quoted_bind_names.get(name, name)] = (
906 out_parameters[name]
907 )
909 def _generate_cursor_outputtype_handler(self):
910 output_handlers = {}
912 for keyname, name, objects, type_ in self.compiled._result_columns:
913 handler = type_._cached_custom_processor(
914 self.dialect,
915 "cx_oracle_outputtypehandler",
916 self._get_cx_oracle_type_handler,
917 )
919 if handler:
920 denormalized_name = self.dialect.denormalize_name(keyname)
921 output_handlers[denormalized_name] = handler
923 if output_handlers:
924 default_handler = self._dbapi_connection.outputtypehandler
926 def output_type_handler(
927 cursor, name, default_type, size, precision, scale
928 ):
929 if name in output_handlers:
930 return output_handlers[name](
931 cursor, name, default_type, size, precision, scale
932 )
933 else:
934 return default_handler(
935 cursor, name, default_type, size, precision, scale
936 )
938 self.cursor.outputtypehandler = output_type_handler
940 def _get_cx_oracle_type_handler(self, impl):
941 if hasattr(impl, "_cx_oracle_outputtypehandler"):
942 return impl._cx_oracle_outputtypehandler(self.dialect)
943 else:
944 return None
946 def pre_exec(self):
947 super().pre_exec()
948 if not getattr(self.compiled, "_oracle_cx_sql_compiler", False):
949 return
951 self.out_parameters = {}
953 self._generate_out_parameter_vars()
955 self._generate_cursor_outputtype_handler()
957 def post_exec(self):
958 if (
959 self.compiled
960 and is_sql_compiler(self.compiled)
961 and self.compiled._oracle_returning
962 ):
963 initial_buffer = self.fetchall_for_returning(
964 self.cursor, _internal=True
965 )
967 fetch_strategy = _cursor.FullyBufferedCursorFetchStrategy(
968 self.cursor,
969 [
970 (entry.keyname, None)
971 for entry in self.compiled._result_columns
972 ],
973 initial_buffer=initial_buffer,
974 )
976 self.cursor_fetch_strategy = fetch_strategy
978 def create_cursor(self):
979 c = self._dbapi_connection.cursor()
980 if self.dialect.arraysize:
981 c.arraysize = self.dialect.arraysize
983 return c
985 def fetchall_for_returning(self, cursor, *, _internal=False):
986 compiled = self.compiled
987 if (
988 not _internal
989 and compiled is None
990 or not is_sql_compiler(compiled)
991 or not compiled._oracle_returning
992 ):
993 raise NotImplementedError(
994 "execution context was not prepared for Oracle RETURNING"
995 )
997 # create a fake cursor result from the out parameters. unlike
998 # get_out_parameter_values(), the result-row handlers here will be
999 # applied at the Result level
1001 numcols = len(self.out_parameters)
1003 # [stmt_result for stmt_result in outparam.values] == each
1004 # statement in executemany
1005 # [val for val in stmt_result] == each row for a particular
1006 # statement
1007 return list(
1008 zip(
1009 *[
1010 [
1011 val
1012 for stmt_result in self.out_parameters[
1013 f"ret_{j}"
1014 ].values
1015 for val in (stmt_result or ())
1016 ]
1017 for j in range(numcols)
1018 ]
1019 )
1020 )
1022 def get_out_parameter_values(self, out_param_names):
1023 # this method should not be called when the compiler has
1024 # RETURNING as we've turned the has_out_parameters flag set to
1025 # False.
1026 assert not self.compiled.returning
1028 return [
1029 self.dialect._paramval(self.out_parameters[name])
1030 for name in out_param_names
1031 ]
1034class OracleDialect_cx_oracle(OracleDialect):
1035 supports_statement_cache = True
1036 execution_ctx_cls = OracleExecutionContext_cx_oracle
1037 statement_compiler = OracleCompiler_cx_oracle
1039 supports_sane_rowcount = True
1040 supports_sane_multi_rowcount = True
1042 insert_executemany_returning = True
1043 insert_executemany_returning_sort_by_parameter_order = True
1044 update_executemany_returning = True
1045 delete_executemany_returning = True
1047 bind_typing = interfaces.BindTyping.SETINPUTSIZES
1049 driver = "cx_oracle"
1051 colspecs = util.update_copy(
1052 OracleDialect.colspecs,
1053 {
1054 sqltypes.TIMESTAMP: _CXOracleTIMESTAMP,
1055 sqltypes.Numeric: _OracleNumeric,
1056 sqltypes.Float: _OracleNumeric,
1057 oracle.BINARY_FLOAT: _OracleBINARY_FLOAT,
1058 oracle.BINARY_DOUBLE: _OracleBINARY_DOUBLE,
1059 sqltypes.Integer: _OracleInteger,
1060 oracle.NUMBER: _OracleNUMBER,
1061 sqltypes.Date: _CXOracleDate,
1062 sqltypes.LargeBinary: _OracleBinary,
1063 sqltypes.Boolean: oracle._OracleBoolean,
1064 sqltypes.Interval: _OracleInterval,
1065 oracle.INTERVAL: _OracleInterval,
1066 sqltypes.Text: _OracleText,
1067 sqltypes.String: _OracleString,
1068 sqltypes.UnicodeText: _OracleUnicodeTextCLOB,
1069 sqltypes.CHAR: _OracleChar,
1070 sqltypes.NCHAR: _OracleNChar,
1071 sqltypes.Enum: _OracleEnum,
1072 oracle.LONG: _OracleLong,
1073 oracle.RAW: _OracleRaw,
1074 sqltypes.Unicode: _OracleUnicodeStringCHAR,
1075 sqltypes.NVARCHAR: _OracleUnicodeStringNCHAR,
1076 sqltypes.Uuid: _OracleUUID,
1077 oracle.NCLOB: _OracleUnicodeTextNCLOB,
1078 oracle.ROWID: _OracleRowid,
1079 },
1080 )
1082 execute_sequence_format = list
1084 _cx_oracle_threaded = None
1086 _cursor_var_unicode_kwargs = util.immutabledict()
1088 @util.deprecated_params(
1089 threaded=(
1090 "1.3",
1091 "The 'threaded' parameter to the cx_oracle/oracledb dialect "
1092 "is deprecated as a dialect-level argument, and will be removed "
1093 "in a future release. As of version 1.3, it defaults to False "
1094 "rather than True. The 'threaded' option can be passed to "
1095 "cx_Oracle directly in the URL query string passed to "
1096 ":func:`_sa.create_engine`.",
1097 )
1098 )
1099 def __init__(
1100 self,
1101 auto_convert_lobs=True,
1102 coerce_to_decimal=True,
1103 arraysize=None,
1104 encoding_errors=None,
1105 threaded=None,
1106 **kwargs,
1107 ):
1108 OracleDialect.__init__(self, **kwargs)
1109 self.arraysize = arraysize
1110 self.encoding_errors = encoding_errors
1111 if encoding_errors:
1112 self._cursor_var_unicode_kwargs = {
1113 "encodingErrors": encoding_errors
1114 }
1115 if threaded is not None:
1116 self._cx_oracle_threaded = threaded
1117 self.auto_convert_lobs = auto_convert_lobs
1118 self.coerce_to_decimal = coerce_to_decimal
1119 if self._use_nchar_for_unicode:
1120 self.colspecs = self.colspecs.copy()
1121 self.colspecs[sqltypes.Unicode] = _OracleUnicodeStringNCHAR
1122 self.colspecs[sqltypes.UnicodeText] = _OracleUnicodeTextNCLOB
1124 dbapi_module = self.dbapi
1125 self._load_version(dbapi_module)
1127 if dbapi_module is not None:
1128 # these constants will first be seen in SQLAlchemy datatypes
1129 # coming from the get_dbapi_type() method. We then
1130 # will place the following types into setinputsizes() calls
1131 # on each statement. Oracle constants that are not in this
1132 # list will not be put into setinputsizes().
1133 self.include_set_input_sizes = {
1134 dbapi_module.DATETIME,
1135 dbapi_module.DB_TYPE_NVARCHAR, # used for CLOB, NCLOB
1136 dbapi_module.DB_TYPE_RAW, # used for BLOB
1137 dbapi_module.NCLOB, # not currently used except for OUT param
1138 dbapi_module.CLOB, # not currently used except for OUT param
1139 dbapi_module.LOB, # not currently used
1140 dbapi_module.BLOB, # not currently used except for OUT param
1141 dbapi_module.NCHAR,
1142 dbapi_module.FIXED_NCHAR,
1143 dbapi_module.FIXED_CHAR,
1144 dbapi_module.TIMESTAMP,
1145 int, # _OracleInteger,
1146 # _OracleBINARY_FLOAT, _OracleBINARY_DOUBLE,
1147 dbapi_module.NATIVE_FLOAT,
1148 }
1150 self._paramval = lambda value: value.getvalue()
1152 def _load_version(self, dbapi_module):
1153 version = (0, 0, 0)
1154 if dbapi_module is not None:
1155 m = re.match(r"(\d+)\.(\d+)(?:\.(\d+))?", dbapi_module.version)
1156 if m:
1157 version = tuple(
1158 int(x) for x in m.group(1, 2, 3) if x is not None
1159 )
1160 self.cx_oracle_ver = version
1161 if self.cx_oracle_ver < (8,) and self.cx_oracle_ver > (0, 0, 0):
1162 raise exc.InvalidRequestError(
1163 "cx_Oracle version 8 and above are supported"
1164 )
1166 @classmethod
1167 def import_dbapi(cls):
1168 import cx_Oracle
1170 return cx_Oracle
1172 def initialize(self, connection):
1173 super().initialize(connection)
1174 self._detect_decimal_char(connection)
1176 def get_isolation_level(self, dbapi_connection):
1177 # sources:
1179 # general idea of transaction id, have to start one, etc.
1180 # https://stackoverflow.com/questions/10711204/how-to-check-isoloation-level
1182 # how to decode xid cols from v$transaction to match
1183 # https://asktom.oracle.com/pls/apex/f?p=100:11:0::::P11_QUESTION_ID:9532779900346079444
1185 # Oracle tuple comparison without using IN:
1186 # https://www.sql-workbench.eu/comparison/tuple_comparison.html
1188 with dbapi_connection.cursor() as cursor:
1189 # this is the only way to ensure a transaction is started without
1190 # actually running DML. There's no way to see the configured
1191 # isolation level without getting it from v$transaction which
1192 # means transaction has to be started.
1193 outval = cursor.var(str)
1194 cursor.execute(
1195 """
1196 begin
1197 :trans_id := dbms_transaction.local_transaction_id( TRUE );
1198 end;
1199 """,
1200 {"trans_id": outval},
1201 )
1202 trans_id = outval.getvalue()
1203 xidusn, xidslot, xidsqn = trans_id.split(".", 2)
1205 cursor.execute(
1206 "SELECT CASE BITAND(t.flag, POWER(2, 28)) "
1207 "WHEN 0 THEN 'READ COMMITTED' "
1208 "ELSE 'SERIALIZABLE' END AS isolation_level "
1209 "FROM v$transaction t WHERE "
1210 "(t.xidusn, t.xidslot, t.xidsqn) = "
1211 "((:xidusn, :xidslot, :xidsqn))",
1212 {"xidusn": xidusn, "xidslot": xidslot, "xidsqn": xidsqn},
1213 )
1214 row = cursor.fetchone()
1215 if row is None:
1216 raise exc.InvalidRequestError(
1217 "could not retrieve isolation level"
1218 )
1219 result = row[0]
1221 return result
1223 def get_isolation_level_values(self, dbapi_connection):
1224 return super().get_isolation_level_values(dbapi_connection) + [
1225 "AUTOCOMMIT"
1226 ]
1228 def set_isolation_level(self, dbapi_connection, level):
1229 if level == "AUTOCOMMIT":
1230 dbapi_connection.autocommit = True
1231 else:
1232 dbapi_connection.autocommit = False
1233 dbapi_connection.rollback()
1234 with dbapi_connection.cursor() as cursor:
1235 cursor.execute(f"ALTER SESSION SET ISOLATION_LEVEL={level}")
1237 def _detect_decimal_char(self, connection):
1238 # we have the option to change this setting upon connect,
1239 # or just look at what it is upon connect and convert.
1240 # to minimize the chance of interference with changes to
1241 # NLS_TERRITORY or formatting behavior of the DB, we opt
1242 # to just look at it
1244 dbapi_connection = connection.connection
1246 with dbapi_connection.cursor() as cursor:
1247 # issue #8744
1248 # nls_session_parameters is not available in some Oracle
1249 # modes like "mount mode". But then, v$nls_parameters is not
1250 # available if the connection doesn't have SYSDBA priv.
1251 #
1252 # simplify the whole thing and just use the method that we were
1253 # doing in the test suite already, selecting a number
1255 def output_type_handler(
1256 cursor, name, defaultType, size, precision, scale
1257 ):
1258 return cursor.var(
1259 self.dbapi.STRING, 255, arraysize=cursor.arraysize
1260 )
1262 cursor.outputtypehandler = output_type_handler
1263 cursor.execute("SELECT 1.1 FROM DUAL")
1264 value = cursor.fetchone()[0]
1266 decimal_char = value.lstrip("0")[1]
1267 assert not decimal_char[0].isdigit()
1269 self._decimal_char = decimal_char
1271 if self._decimal_char != ".":
1272 _detect_decimal = self._detect_decimal
1273 _to_decimal = self._to_decimal
1275 self._detect_decimal = lambda value: _detect_decimal(
1276 value.replace(self._decimal_char, ".")
1277 )
1278 self._to_decimal = lambda value: _to_decimal(
1279 value.replace(self._decimal_char, ".")
1280 )
1282 def _detect_decimal(self, value):
1283 if "." in value:
1284 return self._to_decimal(value)
1285 else:
1286 return int(value)
1288 _to_decimal = decimal.Decimal
1290 def _generate_connection_outputtype_handler(self):
1291 """establish the default outputtypehandler established at the
1292 connection level.
1294 """
1296 dialect = self
1297 cx_Oracle = dialect.dbapi
1299 number_handler = _OracleNUMBER(
1300 asdecimal=True
1301 )._cx_oracle_outputtypehandler(dialect)
1302 float_handler = _OracleNUMBER(
1303 asdecimal=False
1304 )._cx_oracle_outputtypehandler(dialect)
1306 def output_type_handler(
1307 cursor, name, default_type, size, precision, scale
1308 ):
1309 if (
1310 default_type == cx_Oracle.NUMBER
1311 and default_type is not cx_Oracle.NATIVE_FLOAT
1312 ):
1313 if not dialect.coerce_to_decimal:
1314 return None
1315 elif precision == 0 and scale in (0, -127):
1316 # ambiguous type, this occurs when selecting
1317 # numbers from deep subqueries
1318 return cursor.var(
1319 cx_Oracle.STRING,
1320 255,
1321 outconverter=dialect._detect_decimal,
1322 arraysize=cursor.arraysize,
1323 )
1324 elif precision and scale > 0:
1325 return number_handler(
1326 cursor, name, default_type, size, precision, scale
1327 )
1328 else:
1329 return float_handler(
1330 cursor, name, default_type, size, precision, scale
1331 )
1333 # if unicode options were specified, add a decoder, otherwise
1334 # cx_Oracle should return Unicode
1335 elif (
1336 dialect._cursor_var_unicode_kwargs
1337 and default_type
1338 in (
1339 cx_Oracle.STRING,
1340 cx_Oracle.FIXED_CHAR,
1341 )
1342 and default_type is not cx_Oracle.CLOB
1343 and default_type is not cx_Oracle.NCLOB
1344 ):
1345 return cursor.var(
1346 str,
1347 size,
1348 cursor.arraysize,
1349 **dialect._cursor_var_unicode_kwargs,
1350 )
1352 elif dialect.auto_convert_lobs and default_type in (
1353 cx_Oracle.CLOB,
1354 cx_Oracle.NCLOB,
1355 ):
1356 typ = (
1357 cx_Oracle.DB_TYPE_VARCHAR
1358 if default_type is cx_Oracle.CLOB
1359 else cx_Oracle.DB_TYPE_NVARCHAR
1360 )
1361 return cursor.var(
1362 typ,
1363 _CX_ORACLE_MAGIC_LOB_SIZE,
1364 cursor.arraysize,
1365 **dialect._cursor_var_unicode_kwargs,
1366 )
1368 elif dialect.auto_convert_lobs and default_type in (
1369 cx_Oracle.BLOB,
1370 ):
1371 return cursor.var(
1372 cx_Oracle.DB_TYPE_RAW,
1373 _CX_ORACLE_MAGIC_LOB_SIZE,
1374 cursor.arraysize,
1375 )
1377 return output_type_handler
1379 def on_connect(self):
1380 output_type_handler = self._generate_connection_outputtype_handler()
1382 def on_connect(conn):
1383 conn.outputtypehandler = output_type_handler
1385 return on_connect
1387 def create_connect_args(self, url):
1388 opts = dict(url.query)
1390 for opt in ("use_ansi", "auto_convert_lobs"):
1391 if opt in opts:
1392 util.warn_deprecated(
1393 f"{self.driver} dialect option {opt!r} should only be "
1394 "passed to create_engine directly, not within the URL "
1395 "string",
1396 version="1.3",
1397 )
1398 util.coerce_kw_type(opts, opt, bool)
1399 setattr(self, opt, opts.pop(opt))
1401 database = url.database
1402 service_name = opts.pop("service_name", None)
1403 if database or service_name:
1404 # if we have a database, then we have a remote host
1405 port = url.port
1406 if port:
1407 port = int(port)
1408 else:
1409 port = 1521
1411 if database and service_name:
1412 raise exc.InvalidRequestError(
1413 '"service_name" option shouldn\'t '
1414 'be used with a "database" part of the url'
1415 )
1416 if database:
1417 makedsn_kwargs = {"sid": database}
1418 if service_name:
1419 makedsn_kwargs = {"service_name": service_name}
1421 dsn = self.dbapi.makedsn(url.host, port, **makedsn_kwargs)
1422 else:
1423 # we have a local tnsname
1424 dsn = url.host
1426 if dsn is not None:
1427 opts["dsn"] = dsn
1428 if url.password is not None:
1429 opts["password"] = url.password
1430 if url.username is not None:
1431 opts["user"] = url.username
1433 if self._cx_oracle_threaded is not None:
1434 opts.setdefault("threaded", self._cx_oracle_threaded)
1436 def convert_cx_oracle_constant(value):
1437 if isinstance(value, str):
1438 try:
1439 int_val = int(value)
1440 except ValueError:
1441 value = value.upper()
1442 return getattr(self.dbapi, value)
1443 else:
1444 return int_val
1445 else:
1446 return value
1448 util.coerce_kw_type(opts, "mode", convert_cx_oracle_constant)
1449 util.coerce_kw_type(opts, "threaded", bool)
1450 util.coerce_kw_type(opts, "events", bool)
1451 util.coerce_kw_type(opts, "purity", convert_cx_oracle_constant)
1452 return ([], opts)
1454 def _get_server_version_info(self, connection):
1455 return tuple(int(x) for x in connection.connection.version.split("."))
1457 def is_disconnect(self, e, connection, cursor):
1458 (error,) = e.args
1459 if isinstance(
1460 e, (self.dbapi.InterfaceError, self.dbapi.DatabaseError)
1461 ) and "not connected" in str(e):
1462 return True
1464 if hasattr(error, "code") and error.code in {
1465 28,
1466 3114,
1467 3113,
1468 3135,
1469 1033,
1470 2396,
1471 }:
1472 # ORA-00028: your session has been killed
1473 # ORA-03114: not connected to ORACLE
1474 # ORA-03113: end-of-file on communication channel
1475 # ORA-03135: connection lost contact
1476 # ORA-01033: ORACLE initialization or shutdown in progress
1477 # ORA-02396: exceeded maximum idle time, please connect again
1478 # TODO: Others ?
1479 return True
1481 if re.match(r"^(?:DPI-1010|DPI-1080|DPY-1001|DPY-4011)", str(e)):
1482 # DPI-1010: not connected
1483 # DPI-1080: connection was closed by ORA-3113
1484 # python-oracledb's DPY-1001: not connected to database
1485 # python-oracledb's DPY-4011: the database or network closed the
1486 # connection
1487 # TODO: others?
1488 return True
1490 return False
1492 def create_xid(self):
1493 id_ = random.randint(0, 2**128)
1494 return (0x1234, "%032x" % id_, "%032x" % 9)
1496 def do_executemany(self, cursor, statement, parameters, context=None):
1497 if isinstance(parameters, tuple):
1498 parameters = list(parameters)
1499 cursor.executemany(statement, parameters)
1501 def do_begin_twophase(self, connection, xid):
1502 connection.connection.begin(*xid)
1503 connection.connection.info["cx_oracle_xid"] = xid
1505 def do_prepare_twophase(self, connection, xid):
1506 result = connection.connection.prepare()
1507 connection.info["cx_oracle_prepared"] = result
1509 def do_rollback_twophase(
1510 self, connection, xid, is_prepared=True, recover=False
1511 ):
1512 self.do_rollback(connection.connection)
1513 # TODO: need to end XA state here
1515 def do_commit_twophase(
1516 self, connection, xid, is_prepared=True, recover=False
1517 ):
1518 if not is_prepared:
1519 self.do_commit(connection.connection)
1520 else:
1521 if recover:
1522 raise NotImplementedError(
1523 "2pc recovery not implemented for cx_Oracle"
1524 )
1525 oci_prepared = connection.info["cx_oracle_prepared"]
1526 if oci_prepared:
1527 self.do_commit(connection.connection)
1528 # TODO: need to end XA state here
1530 def do_set_input_sizes(self, cursor, list_of_tuples, context):
1531 if self.positional:
1532 # not usually used, here to support if someone is modifying
1533 # the dialect to use positional style
1534 cursor.setinputsizes(
1535 *[dbtype for key, dbtype, sqltype in list_of_tuples]
1536 )
1537 else:
1538 collection = (
1539 (key, dbtype)
1540 for key, dbtype, sqltype in list_of_tuples
1541 if dbtype
1542 )
1544 cursor.setinputsizes(**{key: dbtype for key, dbtype in collection})
1546 def do_recover_twophase(self, connection):
1547 raise NotImplementedError(
1548 "recover two phase query for cx_Oracle not implemented"
1549 )
1552dialect = OracleDialect_cx_oracle