Source code for pydal.adapters.sqlite

import locale
import platform
import re
import sys
from datetime import datetime
from time import mktime
from .._compat import PY2, pjoin
from .base import SQLAdapter
from . import adapters

[docs]@adapters.register_for('sqlite', 'sqlite:memory') class SQLite(SQLAdapter): dbengine = 'sqlite' drivers = ('sqlite2', 'sqlite3') def _initialize_(self, do_connect): self.pool_size = 0 super(SQLite, self)._initialize_(do_connect) path_encoding = sys.getfilesystemencoding() \ or locale.getdefaultlocale()[1] or 'utf8' if ':memory' in self.uri.split('://', 1)[0]: self.dbpath = ':memory:' else: self.dbpath = self.uri.split('://', 1)[1] if self.dbpath[0] != '/': if PY2: self.dbpath = pjoin( self.folder.decode(path_encoding).encode('utf8'), self.dbpath) else: self.dbpath = pjoin(self.folder, self.dbpath) if 'check_same_thread' not in self.driver_args: self.driver_args['check_same_thread'] = False if 'detect_types' not in self.driver_args and do_connect: self.driver_args['detect_types'] = self.driver.PARSE_DECLTYPES def _driver_from_uri(self): return None
[docs] def connector(self): return self.driver.Connection(self.dbpath, **self.driver_args)
[docs] @staticmethod def web2py_extract(lookup, s): table = { 'year': (0, 4), 'month': (5, 7), 'day': (8, 10), 'hour': (11, 13), 'minute': (14, 16), 'second': (17, 19)} try: if lookup != 'epoch': (i, j) = table[lookup] return int(s[i:j]) else: return mktime( datetime.strptime(s, '%Y-%m-%d %H:%M:%S').timetuple()) except: return None
[docs] @staticmethod def web2py_regexp(expression, item): if item is None: return False return re.compile(expression).search(item) is not None
def _register_extract(self): self.connection.create_function( 'web2py_extract', 2, self.web2py_extract) def _register_regexp(self): self.connection.create_function( "REGEXP", 2, self.web2py_regexp)
[docs] def after_connection(self): self._register_extract() self._register_regexp() if self.adapter_args.get('foreign_keys', True): self.execute('PRAGMA foreign_keys=ON;')
[docs] def select(self, query, fields, attributes): if attributes.get('for_update', False) and 'cache' not in attributes: self.execute('BEGIN IMMEDIATE TRANSACTION;') return super(SQLite, self).select(query, fields, attributes)
[docs] def delete(self, table, query): db = self.db deleted = [x[] for x in db(query).select(table._id)] counter = super(SQLite, self).delete(table, query) if counter: for field in table._referenced_by: if field.type == 'reference ' + table._dalname \ and field.ondelete == 'CASCADE': db(field.belongs(deleted)).delete() return counter
[docs]@adapters.register_for('spatialite', 'spatialite:memory') class Spatialite(SQLite): dbengine = 'spatialite' SPATIALLIBS = { 'Windows': 'mod_spatialite.dll', 'Linux': '', 'Darwin': 'libspatialite.dylib' }
[docs] def after_connections(self): self.connection.enable_load_extension(True) libspatialite = self.SPATIALLIBS[platform.system()] self.execute(r'SELECT load_extension("%s");' % libspatialite) super(Spatialite, self).after_connection()
[docs]@adapters.register_for('jdbc:sqlite', 'jdbc:sqlite:memory') class JDBCSQLite(SQLite): drivers = ('zxJDBC_sqlite',)
[docs] def connector(self): return self.driver.connect( self.driver.getConnection('jdbc:sqlite:'+self.dbpath), **self.driver_args)
[docs] def after_connection(self): self._register_extract()