htsql / src / htsql_oracle / core /

# Copyright (c) 2006-2013, Prometheus Research, LLC

from htsql.core.connect import Connect, Scramble, Unscramble, UnscrambleError
from htsql.core.adapter import adapt
from htsql.core.context import context
from htsql.core.error import Error
from htsql.core.domain import (BooleanDomain, DecimalDomain, TextDomain,
        DateDomain, TimeDomain)
import datetime
import decimal
import cx_Oracle

class ConnectOracle(Connect):
    Implementation of the connection adapter for Oracle.

    def outconverter(cls, value):
        value = value.replace(',', '.')
        if '.' in value:
            return decimal.Decimal(value)
        return int(value)

    def outputtypehandler(cls, cursor, name, defaultType,
                          size, precision, scale):
        if defaultType == cx_Oracle.NUMBER:
            return cursor.var(str, 100, cursor.arraysize,
        if defaultType in (cx_Oracle.STRING, cx_Oracle.FIXED_CHAR):
            return cursor.var(unicode, size, cursor.arraysize)

    def open(self):
        addon =
        parameters = {}
        parameters['user'] = addon.db.username or ''
        parameters['password'] = addon.db.password or ''
        if addon.password is not None:
            parameters['password'] = addon.password
        if is not None:
            host =
            port = addon.db.port or 1521
            sid = addon.db.database
            dsn = cx_Oracle.makedsn(host, port, sid)
            dsn = addon.db.database
        parameters['dsn'] = dsn
        connection = cx_Oracle.connect(**parameters)
        if self.with_autocommit:
            connection.autocommit = True
        connection.outputtypehandler = self.outputtypehandler
        cursor = connection.cursor()
        cursor.execute("ALTER SESSION SET NLS_SORT = BINARY_CI")
        cursor.execute("ALTER SESSION SET NLS_COMP = LINGUISTIC")
        return connection

class UnscrambleOracleError(UnscrambleError):

    def __call__(self):
        # If we got a DBAPI exception, extract the error message.
        if isinstance(self.error, cx_Oracle.Error):
            return str(self.error)

        # Otherwise, let the superclass return `None`.
        return super(UnscrambleOracleError, self).__call__()

class UnscrambleOracleBoolean(Unscramble):


    def convert(value):
        if value is None:
            return None
        return (value != 0)

class UnscrambleOracleDecimal(Unscramble):


    def convert(value):
        if value is None:
            return None
        return decimal.Decimal(value)

class UnscrambleOracleText(Unscramble):


    def convert(value):
        if isinstance(value, cx_Oracle.LOB):
                value =
            except cx_Oracle.Error, exc:
                message = str(exc)
                raise Error(message, exc)
        if isinstance(value, str):
            value = value.decode('utf-8')
        return value

class UnscrambleOracleDate(Unscramble):


    def convert(value):
        if isinstance(value, datetime.datetime):
            assert not value.time()
            value =
        return value

class UnscrambleOracleTime(Unscramble):


    def convert(value):
        if isinstance(value, datetime.timedelta):
            assert not value.days
            value = (datetime.datetime(2001,1,1) + value).time()
        return value
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.