""" $URL: svn+ssh://svn.mikewatkins.ca/repo/trunk/parlez/lib/dbapi_storage.py $ $Id: dbapi_storage.py 41 2007-09-02 14:08:04Z mw $ Provides an abstract DBAPIStorage; two working implementations are provided PostgresqlStorage (supporting psycopg2 driver) and SqliteStorage (supporting Python 2.5's built in sqlite3 DBAPI interface as well as pysqlite2). Motivation to do this thanks to a sqlite/apsw specific effort by Peter Wilkinson of Thirdfloor Software Works Pty. Ltd. """ from durus.serialize import unpack_record from durus.connection import ROOT_OID from durus.utils import str_to_int8 from durus.storage import Storage from itertools import islice, chain class DBAPIStorage(Storage): """ This is an abstract class which SQL-based Durus storages can inherit. This class assumes a Python DB API compatible connection. """ def __init__(self, sql_connection, table_name='durus_records', **kwargs): if self.__class__ is DBAPIStorage: raise RuntimeError("DBAPIStorage is abstract") self.ROOT_OID = ROOT_OID self._sql_connection = sql_connection self._sql_table_name = table_name self._sql_options = dict( pack_increment = 1000, ) self._sql_options.update(kwargs) self._sql_ensure_tables() self.transaction = [] def _sql_execute(self, sql, *args): """(sql:str, *args) -> DBAPI cursor object All DBAPI adapters behave differently upon a cursor.execute() - some return the cursor object, some do not. Some return a logical True or False value. Here we return a cursor regardless of dbapi adapter. """ cursor = self._sql_connection.cursor() cursor.execute(sql, *args) return cursor def _sql_queryone(self, sql, *args): """(sql:str, *args) -> SQL query result or None Encapsulates cursor(); execute(); fetchone(). Not all queries return results (i.e. CREATE TABLE, UPDATE, INSERT), and DBAPI adapters behave differently on calls to fetchone() - some return a ProgrammingError (or other such exception) if there are no results. Some return None. Use _sql_execute() if you expect no results. This method will not trap errors raised by the DBAPI layer if there are no results to return. """ cursor = self._sql_execute(sql, *args) return cursor.fetchone() def _sql_ensure_tables(self): """ Ensure table(s) required exist, create if not. Table existence detection: Postgres: SELECT tablename from pg_tables where schemaname='public' AND tablename= sqlite: PRAGMA table_info() """ def tables_exist(): pass def create_tables(): pass if not tables_exist(): create_tables() raise NotImplementedError, ( 'Table existence detection and creation is not portable, override ' 'in your subclass.') def new_oid(self): """() -> str Must return return int8_to_str() """ raise NotImplementedError, ('OID generation is not portable') def load(self, oid): """(oid:str) -> record:str One of the few portable queries and can be used as-is if you are willing to live without using a proper bind variable. You might also want to override if the DPAPI adapter being used happens to provide some enhancement that makes sense to take advantage of, such as prepared statements. """ result = self._sql_queryone( 'SELECT record FROM %s WHERE p_oid = %s' % (self._sql_table_name, str_to_int8(oid))) if result: return str(result[0]) else: raise KeyError, 'oid %s not in storage' % str_to_int8(oid) def begin(self): """() -> None Begin a new Durus transaction (a commit). """ self.transaction = {} def store(self, oid, record): """(oid:str, record:object) -> None Adds record to the transaction queue for a commit underway. """ assert len(oid) == 8 assert oid not in self.transaction self.transaction[oid] = record def end(self, handle_invalidations=None): """ Conclude a commit. """ raise NotImplementedError, ('Parameter passing to queries not portable') def sync(self): return [] def close(self): """Close storage database connections""" if self._sql_connection: self._sql_connection.close() def get_packer(self): """ Returns an incremental packer used by StorageServer. This implementation of _packer can be used for any DBAPI driver as standard python string parameters and dbapi cursors are used. """ def gen_oids(): for oid, record in self.gen_oid_record(batch_size=pack_increment): record_oid, data, refdata = unpack_record(record) assert oid == record_oid yield str_to_int8(oid) def gen_batch(iterable, size=None): iterable = iter(iterable) while True: chunk = islice(iterable, size) yield chain([chunk.next()], chunk) def packer(): assert not self.transaction self._sql_execute('BEGIN') for batch in gen_batch(gen_oids(), size=pack_increment): # this is significantly faster than executemany sql = "UPDATE %s set pack = 1 WHERE p_oid in (%s)" %\ (self._sql_table_name, ','.join(map(str,batch))) self._sql_execute(sql) yield None self._sql_connection.execute( "DELETE FROM %s WHERE pack = 0" % self._sql_table_name) self._sql_connection.execute( "UPDATE %s SET pack = 0" % self._sql_table_name) self._sql_execute('COMMIT') pack_increment = self._sql_options.get('pack_increment', 1000) return packer() def pack(self): """ Remove obsolete records. """ for z in self.get_packer(): pass def get_size(self): """() -> int Many SQL databases impose a large cost on select count(*) from ... thus this is not implemented, returning None. Subclasses may have db specific performance workarounds to offer... """ return None def bulk_load(self, oids): """(oids:sequence(oid:str)) -> sequence(record:str) This portable approach is approx 3X faster than loading one by one. """ requested_oids = map(str_to_int8, oids) requested_oids.sort() cursor = self._sql_connection.cursor() cursor.execute( """ SELECT p_oid, record from %s WHERE p_oid in (%s) AND record IS NOT NULL """ %\ (self._sql_table_name, ','.join(map(str, requested_oids)))) result = cursor.fetchall() returned_oids = [int_oid for int_oid, record in result] returned_oids.sort() if requested_oids != returned_oids: # shouldn't occur under normal circumstances not_found = [] for r_oid in requested_oids: if r_oid not in returned_oids: not_found += [r_oid] raise KeyError('oids %s not in DBAPIStorage' % ', '.join(map(str, not_found))) for p_oid, record in result: yield record def gen_oid_record(self, start_oid=None, batch_size=None): # use default algorithm but take advantage of larger batch size # which is sensible for a SQL storage # note that this won't guarantee all bulk_load() requests are # batch_size in length; the traversal of the object graph # will determine size. For large collections within lists or mappings # a larger batch size should help if not batch_size: batch_size = self._sql_options.get('pack_increment', 1000) return Storage.gen_oid_record(self, start_oid=start_oid, batch_size=batch_size)