Server IP : 85.214.239.14 / Your IP : 18.117.254.177 Web Server : Apache/2.4.62 (Debian) System : Linux h2886529.stratoserver.net 4.9.0 #1 SMP Tue Jan 9 19:45:01 MSK 2024 x86_64 User : www-data ( 33) PHP Version : 7.4.18 Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare, MySQL : OFF | cURL : OFF | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : OFF Directory : /proc/self/root/srv/automx/env/lib64/python3.5/site-packages/sqlalchemy/dialects/oracle/ |
Upload File : |
# oracle/zxjdbc.py # Copyright (C) 2005-2020 the SQLAlchemy authors and contributors # <see AUTHORS file> # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php """ .. dialect:: oracle+zxjdbc :name: zxJDBC for Jython :dbapi: zxjdbc :connectstring: oracle+zxjdbc://user:pass@host/dbname :driverurl: http://www.oracle.com/technetwork/database/features/jdbc/index-091264.html .. note:: Jython is not supported by current versions of SQLAlchemy. The zxjdbc dialect should be considered as experimental. """ # noqa import collections import decimal import re from .base import OracleCompiler from .base import OracleDialect from .base import OracleExecutionContext from ... import sql from ... import types as sqltypes from ... import util from ...connectors.zxJDBC import ZxJDBCConnector from ...engine import result as _result from ...sql import expression SQLException = zxJDBC = None class _ZxJDBCDate(sqltypes.Date): def result_processor(self, dialect, coltype): def process(value): if value is None: return None else: return value.date() return process class _ZxJDBCNumeric(sqltypes.Numeric): def result_processor(self, dialect, coltype): # XXX: does the dialect return Decimal or not??? # if it does (in all cases), we could use a None processor as well as # the to_float generic processor if self.asdecimal: def process(value): if isinstance(value, decimal.Decimal): return value else: return decimal.Decimal(str(value)) else: def process(value): if isinstance(value, decimal.Decimal): return float(value) else: return value return process class OracleCompiler_zxjdbc(OracleCompiler): def returning_clause(self, stmt, returning_cols): self.returning_cols = list( expression._select_iterables(returning_cols) ) # within_columns_clause=False so that labels (foo AS bar) don't render columns = [ self.process(c, within_columns_clause=False) for c in self.returning_cols ] if not hasattr(self, "returning_parameters"): self.returning_parameters = [] binds = [] for i, col in enumerate(self.returning_cols): dbtype = col.type.dialect_impl(self.dialect).get_dbapi_type( self.dialect.dbapi ) self.returning_parameters.append((i + 1, dbtype)) bindparam = sql.bindparam( "ret_%d" % i, value=ReturningParam(dbtype) ) self.binds[bindparam.key] = bindparam binds.append( self.bindparam_string(self._truncate_bindparam(bindparam)) ) return "RETURNING " + ", ".join(columns) + " INTO " + ", ".join(binds) class OracleExecutionContext_zxjdbc(OracleExecutionContext): def pre_exec(self): if hasattr(self.compiled, "returning_parameters"): # prepare a zxJDBC statement so we can grab its underlying # OraclePreparedStatement's getReturnResultSet later self.statement = self.cursor.prepare(self.statement) def get_result_proxy(self): if hasattr(self.compiled, "returning_parameters"): rrs = None try: try: rrs = self.statement.__statement__.getReturnResultSet() next(rrs) except SQLException as sqle: msg = "%s [SQLCode: %d]" % ( sqle.getMessage(), sqle.getErrorCode(), ) if sqle.getSQLState() is not None: msg += " [SQLState: %s]" % sqle.getSQLState() raise zxJDBC.Error(msg) else: row = tuple( self.cursor.datahandler.getPyObject(rrs, index, dbtype) for index, dbtype in self.compiled.returning_parameters ) return ReturningResultProxy(self, row) finally: if rrs is not None: try: rrs.close() except SQLException: pass self.statement.close() return _result.ResultProxy(self) def create_cursor(self): cursor = self._dbapi_connection.cursor() cursor.datahandler = self.dialect.DataHandler(cursor.datahandler) return cursor class ReturningResultProxy(_result.FullyBufferedResultProxy): """ResultProxy backed by the RETURNING ResultSet results.""" def __init__(self, context, returning_row): self._returning_row = returning_row super(ReturningResultProxy, self).__init__(context) def _cursor_description(self): ret = [] for c in self.context.compiled.returning_cols: if hasattr(c, "name"): ret.append((c.name, c.type)) else: ret.append((c.anon_label, c.type)) return ret def _buffer_rows(self): return collections.deque([self._returning_row]) class ReturningParam(object): """A bindparam value representing a RETURNING parameter. Specially handled by OracleReturningDataHandler. """ def __init__(self, type_): self.type = type_ def __eq__(self, other): if isinstance(other, ReturningParam): return self.type == other.type return NotImplemented def __ne__(self, other): if isinstance(other, ReturningParam): return self.type != other.type return NotImplemented def __repr__(self): kls = self.__class__ return "<%s.%s object at 0x%x type=%s>" % ( kls.__module__, kls.__name__, id(self), self.type, ) class OracleDialect_zxjdbc(ZxJDBCConnector, OracleDialect): jdbc_db_name = "oracle" jdbc_driver_name = "oracle.jdbc.OracleDriver" statement_compiler = OracleCompiler_zxjdbc execution_ctx_cls = OracleExecutionContext_zxjdbc colspecs = util.update_copy( OracleDialect.colspecs, {sqltypes.Date: _ZxJDBCDate, sqltypes.Numeric: _ZxJDBCNumeric}, ) def __init__(self, *args, **kwargs): super(OracleDialect_zxjdbc, self).__init__(*args, **kwargs) global SQLException, zxJDBC from java.sql import SQLException from com.ziclix.python.sql import zxJDBC from com.ziclix.python.sql.handler import OracleDataHandler class OracleReturningDataHandler(OracleDataHandler): """zxJDBC DataHandler that specially handles ReturningParam.""" def setJDBCObject(self, statement, index, object_, dbtype=None): if type(object_) is ReturningParam: statement.registerReturnParameter(index, object_.type) elif dbtype is None: OracleDataHandler.setJDBCObject( self, statement, index, object_ ) else: OracleDataHandler.setJDBCObject( self, statement, index, object_, dbtype ) self.DataHandler = OracleReturningDataHandler def initialize(self, connection): super(OracleDialect_zxjdbc, self).initialize(connection) self.implicit_returning = connection.connection.driverversion >= "10.2" def _create_jdbc_url(self, url): return "jdbc:oracle:thin:@%s:%s:%s" % ( url.host, url.port or 1521, url.database, ) def _get_server_version_info(self, connection): version = re.search( r"Release ([\d\.]+)", connection.connection.dbversion ).group(1) return tuple(int(x) for x in version.split(".")) dialect = OracleDialect_zxjdbc