# $SnapHashLicense:
#
# SnapLogic - Open source data services
#
# Copyright (C) 2008-2009, SnapLogic, Inc. All rights reserved.
#
# See http://www.snaplogic.org for more information about
# the SnapLogic project.
#
# This program is free software, distributed under the terms of
# the GNU General Public License Version 2. See the LEGAL file
# at the top of the source tree.
#
# "SnapLogic" is a trademark of SnapLogic, Inc.
#
#
# $
# $Id: SQLite.py 8009 2009-06-26 04:30:52Z grisha $
from decimal import Decimal
import sqlite3 as sqlite
from snaplogic.components.DBUtils import SnapDBAdapter
from snaplogic.common.snap_exceptions import SnapException,SnapComponentError
class SqliteCursorWrapper(object):
"""
Implementation of L{SnapDBAdapter} for Sqlite.
"""
def __init__(self, cursor):
self._delegate = cursor
self._metadata = None
self._string_fields = None
def convert_row(self, row):
"""
Convert a row of data in native data types into a row of Snap types.
@param row: row returned by database
@type row: tuple
@return: row converted to Snap data types
@rtype: list
"""
row = list(row)
for i in range(len(row)):
field = row[i]
if field is None:
continue
t = type(field)
# No metadata is being sent, unfortunately
# So we have to convert by hand
if t == int or t == long:
row[i] = Decimal(field)
elif t == float:
row[i] = Decimal(str(field))
return row
def convert_results(self, rs):
"""
Convert the result set from native data types to Snap data types.
This is similar to L{convert_row}, except it acts on the entire result
set
@param rs: Result set to convert
@type rs: list or tuple
@return: converted result set
@type: list
"""
rs = list(rs)
for i in range(len(rs)):
rs[i] = self.convert_row(rs[i])
return rs
def fetchone(self):
"""
Same as cursor.fetchone() specified in DB API 2.0, except returning
Snap data types.
"""
row = self._delegate.fetchone()
if row is not None:
row = self.convert_row(row)
return row
def fetchmany(self, size=None):
"""
Same as cursor.fetchmany() specified in DB API 2.0, except returning
Snap data types.
"""
if size is None:
size = self._delegate.arraysize
rs = self._delegate.fetchmany(size)
rs = self.convert_results(rs)
return rs
def fetchall(self):
"""
Same as cursor.fetchall() specified in DB API 2.0, except returning
Snap data types.
"""
rs = self._delegate.fetchall()
rs = self.convert_results(rs)
return rs
def __getattr__(self, name):
"""
Used to delegate to the native cursor object those methods that are not
wrapped by this class.
"""
result = getattr(self._delegate, name)
return result
class SQLite(SnapDBAdapter):
def __init__(self, *args, **kwargs):
conn = sqlite.connect(kwargs['path'])
super(SQLite, self).__init__(conn, sqlite)
def cursor(self):
native_cursor = SnapDBAdapter.cursor(self)
my_cursor = SqliteCursorWrapper(native_cursor)
return my_cursor
def bindValueContainer(self, record):
"""
Overrides L{SnapDBAdapter.bindValueContainer} and properly encodes
the parameters.
"""
result = SnapDBAdapter.bindValueContainer(self, record)
new_result = []
for value in result:
value_t = type(value)
if value_t == Decimal:
# TODO!
# Somehow sqlite is not liking this...
int_val = value.__int__()
if value.__eq__(int_val):
value = int_val
else:
value = value.__float__()
new_result.append(value)
return new_result
def get_no_row_query(self, select_sql):
raise SnapComponentError('Operation not supported for this DBMS')
def get_snap_view_metadata(self, select_sql):
raise SnapComponentError('Operation not supported for this DBMS')
|