import base64 import os try: from _thread import get_ident except ImportError: from thread import get_ident from threading import enumerate as enumerate_threads try: from urllib import quote except ImportError: from urllib.parse import quote from sqlobject import col from sqlobject import dberrors from sqlobject.dbconnection import DBAPI, Boolean sqlite2_Binary = None class ErrorMessage(str): def __new__(cls, e): obj = str.__new__(cls, e.args[0]) obj.code = None obj.module = e.__module__ obj.exception = e.__class__.__name__ return obj class SQLiteConnection(DBAPI): supportTransactions = True dbName = 'sqlite' schemes = [dbName] def __init__(self, filename, autoCommit=1, **kw): import sqlite3 as sqlite sqlite.encode = base64.b64encode sqlite.decode = base64.b64decode self.module = sqlite self.filename = filename # full path to sqlite-db-file self._memory = filename == ':memory:' # connection options opts = {} if autoCommit: opts["isolation_level"] = None global sqlite2_Binary if sqlite2_Binary is None: sqlite2_Binary = sqlite.Binary sqlite.Binary = lambda s: sqlite2_Binary(sqlite.encode(s)) if 'factory' in kw: factory = kw.pop('factory') if isinstance(factory, str): factory = globals()[factory] opts['factory'] = factory(sqlite) if 'timeout' in kw: opts['timeout'] = float(kw.pop('timeout')) if 'check_same_thread' in kw: opts["check_same_thread"] = Boolean(kw.pop('check_same_thread')) # use only one connection for sqlite - supports multiple) # cursors per connection self._connOptions = opts self.use_table_info = Boolean(kw.pop("use_table_info", True)) DBAPI.__init__(self, **kw) self._threadPool = {} self._threadOrigination = {} if self._memory: self.makeMemoryConnection() @classmethod def _connectionFromParams(cls, user, password, host, port, path, args): assert host is None and port is None, ( "SQLite can only be used locally (with a URI like " "sqlite:/file or sqlite:///file, not sqlite://%s%s)" % (host, port and ':%r' % port or '')) assert user is None and password is None, ( "You may not provide usernames or passwords for SQLite " "databases") if path == "/:memory:": path = ":memory:" return cls(filename=path, **args) def oldUri(self): path = self.filename if path == ":memory:": path = "/:memory:" else: path = "//" + path return 'sqlite:%s' % path def uri(self): path = self.filename if path == ":memory:": path = "/:memory:" else: if path.startswith('/'): path = "//" + path else: path = "///" + path path = quote(path) return 'sqlite:%s' % path def getConnection(self): # SQLite can't share connections between threads, and so can't # pool connections. Since we are isolating threads here, we # don't have to worry about locking as much. if self._memory: conn = self.makeConnection() self._connectionNumbers[id(conn)] = self._connectionCount self._connectionCount += 1 return conn self._releaseUnusedConnections() threadid = get_ident() if (self._pool is not None and threadid in self._threadPool): conn = self._threadPool[threadid] del self._threadPool[threadid] if conn in self._pool: self._pool.remove(conn) else: conn = self.makeConnection() if self._pool is not None: self._threadOrigination[id(conn)] = threadid self._connectionNumbers[id(conn)] = self._connectionCount self._connectionCount += 1 if self.debug: s = 'ACQUIRE' if self._pool is not None: s += ' pool=[%s]' % ', '.join( [str(self._connectionNumbers[id(v)]) for v in self._pool]) self.printDebug(conn, s, 'Pool') return conn def releaseConnection(self, conn, explicit=False): if self._memory: return threadid = self._threadOrigination.get(id(conn)) DBAPI.releaseConnection(self, conn, explicit=explicit) if (self._pool is not None and threadid and threadid not in self._threadPool): self._threadPool[threadid] = conn else: if self._pool and conn in self._pool: self._pool.remove(conn) if threadid: del self._threadOrigination[id(conn)] del self._threadPool[threadid] conn.close() def _setAutoCommit(self, conn, auto): if auto: conn.isolation_level = None else: conn.isolation_level = "" def _setIsolationLevel(self, conn, level): conn.isolation_level = level def makeMemoryConnection(self): self._memoryConn = self.module.connect( self.filename, **self._connOptions) # Convert text data from SQLite to str, not unicode - # SQLObject converts it to unicode itself. self._memoryConn.text_factory = str def makeConnection(self): if self._memory: return self._memoryConn conn = self.module.connect(self.filename, **self._connOptions) conn.text_factory = str # Convert text data to str, not unicode return conn def _releaseUnusedConnections(self): """Release connections from threads that're no longer active""" thread_ids = set(t.ident for t in enumerate_threads()) tp_set = set(self._threadPool) unused_connections = [ self._threadPool[tid] for tid in tp_set - thread_ids ] for unused_connection in unused_connections: try: self.releaseConnection(unused_connection, explicit=True) except self.module.ProgrammingError: pass # Ignore error in `conn.close()` from a different thread def close(self): DBAPI.close(self) self._threadPool = {} self._threadOrigination = {} if self._memory: self._memoryConn.close() self.makeMemoryConnection() def _executeRetry(self, conn, cursor, query): if self.debug: self.printDebug(conn, query, 'QueryR') try: return cursor.execute(query) except self.module.OperationalError as e: raise dberrors.OperationalError(ErrorMessage(e)) except self.module.IntegrityError as e: msg = ErrorMessage(e) if msg.startswith('column') and msg.endswith('not unique') \ or msg.startswith('UNIQUE constraint failed:'): raise dberrors.DuplicateEntryError(msg) else: raise dberrors.IntegrityError(msg) except self.module.InternalError as e: raise dberrors.InternalError(ErrorMessage(e)) except self.module.ProgrammingError as e: raise dberrors.ProgrammingError(ErrorMessage(e)) except self.module.DataError as e: raise dberrors.DataError(ErrorMessage(e)) except self.module.NotSupportedError as e: raise dberrors.NotSupportedError(ErrorMessage(e)) except self.module.DatabaseError as e: raise dberrors.DatabaseError(ErrorMessage(e)) except self.module.InterfaceError as e: raise dberrors.InterfaceError(ErrorMessage(e)) except self.module.Warning as e: raise Warning(ErrorMessage(e)) except self.module.Error as e: raise dberrors.Error(ErrorMessage(e)) def _queryInsertID(self, conn, soInstance, id, names, values): table = soInstance.sqlmeta.table idName = soInstance.sqlmeta.idName c = conn.cursor() if id is not None: names = [idName] + names values = [id] + values q = self._insertSQL(table, names, values) if self.debug: self.printDebug(conn, q, 'QueryIns') self._executeRetry(conn, c, q) # lastrowid is a DB-API extension from "PEP 0249": if id is None: id = int(c.lastrowid) c.close() if self.debugOutput: self.printDebug(conn, id, 'QueryIns', 'result') return id def _insertSQL(self, table, names, values): if not names: assert not values # INSERT INTO table () VALUES () isn't allowed in # SQLite (though it is in other databases) return ("INSERT INTO %s VALUES (NULL)" % table) else: return DBAPI._insertSQL(self, table, names, values) @classmethod def _queryAddLimitOffset(cls, query, start, end): if not start: return "%s LIMIT %i" % (query, end) if not end: return "%s LIMIT 0 OFFSET %i" % (query, start) return "%s LIMIT %i OFFSET %i" % (query, end - start, start) def createColumn(self, soClass, col): return col.sqliteCreateSQL() def createReferenceConstraint(self, soClass, col): return None def createIDColumn(self, soClass): return self._createIDColumn(soClass.sqlmeta) def _createIDColumn(self, sqlmeta): if sqlmeta.idType is str: return '%s TEXT PRIMARY KEY' % sqlmeta.idName if sqlmeta.idType is not int: raise TypeError('sqlmeta.idType must be int or str, not %r' % sqlmeta.idType) return '%s INTEGER PRIMARY KEY AUTOINCREMENT' % sqlmeta.idName def joinSQLType(self, join): return 'INT NOT NULL' def tableExists(self, tableName): result = self.queryOne( "SELECT tbl_name FROM sqlite_master " "WHERE type='table' AND tbl_name = '%s'" % tableName) # turn it into a boolean: return not not result def createIndexSQL(self, soClass, index): return index.sqliteCreateIndexSQL(soClass) def addColumn(self, tableName, column): self.query('ALTER TABLE %s ADD COLUMN %s' % (tableName, column.sqliteCreateSQL())) self.query('VACUUM') def delColumn(self, sqlmeta, column): self.recreateTableWithoutColumn(sqlmeta, column) def recreateTableWithoutColumn(self, sqlmeta, column): new_name = sqlmeta.table + '_ORIGINAL' self.query('ALTER TABLE %s RENAME TO %s' % (sqlmeta.table, new_name)) cols = [self._createIDColumn(sqlmeta)] + \ [self.createColumn(None, col) for col in sqlmeta.columnList if col.name != column.name] cols = ",\n".join([" %s" % c for c in cols]) self.query('CREATE TABLE %s (\n%s\n)' % (sqlmeta.table, cols)) all_columns = ', '.join( [sqlmeta.idName] + [col.dbName for col in sqlmeta.columnList]) self.query('INSERT INTO %s (%s) SELECT %s FROM %s' % ( sqlmeta.table, all_columns, all_columns, new_name)) self.query('DROP TABLE %s' % new_name) def columnsFromSchema(self, tableName, soClass): if self.use_table_info: return self._columnsFromSchemaTableInfo(tableName, soClass) else: return self._columnsFromSchemaParse(tableName, soClass) def _columnsFromSchemaTableInfo(self, tableName, soClass): colData = self.queryAll("PRAGMA table_info(%s)" % tableName) results = [] for index, field, t, nullAllowed, default, key in colData: if field == soClass.sqlmeta.idName: continue colClass, kw = self.guessClass(t) if default == 'NULL': nullAllowed = True default = None kw['name'] = soClass.sqlmeta.style.dbColumnToPythonAttr(field) kw['dbName'] = field kw['notNone'] = not nullAllowed kw['default'] = default # @@ skip key... # @@ skip extra... results.append(colClass(**kw)) return results def _columnsFromSchemaParse(self, tableName, soClass): colData = self.queryOne( "SELECT sql FROM sqlite_master WHERE type='table' AND name='%s'" % tableName) if not colData: raise ValueError( 'The table %s was not found in the database. Load failed.' % tableName) colData = colData[0].split('(', 1)[1].strip()[:-2] while True: start = colData.find('(') if start == -1: break end = colData.find(')', start) if end == -1: break colData = colData[:start] + colData[end + 1:] results = [] for colDesc in colData.split(','): parts = colDesc.strip().split(' ', 2) field = parts[0].strip() # skip comments if field.startswith('--'): continue # get rid of enclosing quotes if field[0] == field[-1] == '"': field = field[1:-1] if field == getattr(soClass.sqlmeta, 'idName', 'id'): continue colClass, kw = self.guessClass(parts[1].strip()) if len(parts) == 2: index_info = '' else: index_info = parts[2].strip().upper() kw['name'] = soClass.sqlmeta.style.dbColumnToPythonAttr(field) kw['dbName'] = field import re nullble = re.search(r'(\b\S*)\sNULL', index_info) default = re.search( r"DEFAULT\s((?:\d[\dA-FX.]*)|(?:'[^']*')|(?:#[^#]*#))", index_info) kw['notNone'] = nullble and nullble.group(1) == 'NOT' kw['default'] = default and default.group(1) # @@ skip key... # @@ skip extra... results.append(colClass(**kw)) return results def guessClass(self, t): t = t.upper() if t.find('INT') >= 0: return col.IntCol, {} elif t.find('TEXT') >= 0 or t.find('CHAR') >= 0 or t.find('CLOB') >= 0: return col.StringCol, {'length': 2 ** 32 - 1} elif t.find('BLOB') >= 0: return col.BLOBCol, {"length": 2 ** 32 - 1} elif t.find('REAL') >= 0 or t.find('FLOAT') >= 0: return col.FloatCol, {} elif t.find('DECIMAL') >= 0: return col.DecimalCol, {'size': None, 'precision': None} elif t.find('BOOL') >= 0: return col.BoolCol, {} else: return col.Col, {} def listTables(self): return [v[0] for v in self.queryAll( "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name")] def listDatabases(self): # The pragma returns a list of (index, name, filename) return [v[1] for v in self.queryAll("PRAGMA database_list")] def createEmptyDatabase(self): if self._memory: return open(self.filename, 'w').close() def dropDatabase(self): if self._memory: return os.unlink(self.filename)