Source

htsql / src / htsql_oracle / core / connect.py

#
# 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.
    """

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

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

    def open(self):
        addon = context.app.htsql
        parameters = {}
        parameters['user'] = addon.db.username or ''
        parameters['password'] = addon.db.password or ''
        if addon.password is not None:
            parameters['password'] = addon.password
        if addon.db.host is not None:
            host = addon.db.host
            port = addon.db.port or 1521
            sid = addon.db.database
            dsn = cx_Oracle.makedsn(host, port, sid)
        else:
            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):

    adapt(BooleanDomain)

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


class UnscrambleOracleDecimal(Unscramble):

    adapt(DecimalDomain)

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


class UnscrambleOracleText(Unscramble):

    adapt(TextDomain)

    @staticmethod
    def convert(value):
        if isinstance(value, cx_Oracle.LOB):
            try:
                value = value.read()
            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):

    adapt(DateDomain)

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


class UnscrambleOracleTime(Unscramble):

    adapt(TimeDomain)

    @staticmethod
    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 ProjectModifiedEvent.java.
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.