<commit_before># -*- test-case-name: axiom.test -*-

import time
import os
import itertools

from zope.interface import implements

from twisted.python.filepath import FilePath
from twisted.internet import defer
from twisted.python.reflect import qual
from twisted.application.service import IService, MultiService

from axiom import _schema, attributes, upgrade, _fincache, iaxiom

from pysqlite2 import dbapi2 as sqlite

from axiom.item import Item, \
    _typeNameToMostRecentClass, dummyItemSubclass,\
    _legacyTypes, TABLE_NAME, Empowered

IN_MEMORY_DATABASE = ':memory:'

tempCounter = itertools.count()

class XFilePath(FilePath):
    def dirname(self):
        return os.path.dirname(self.path)

def _md(dirname):
    if os.path.isdir(dirname):
        return False
    os.makedirs(dirname)
    return True

class AtomicFile(file):
    """I am a file which is moved from temporary to permanent storage when it
    is closed.

    After I'm closed, I will have a 'finalpath' property saying where I went.
    """

    implements(iaxiom.IAtomicFile)

    def __init__(self, tempname, destpath):
        self._destpath = destpath
        file.__init__(self, tempname, 'w+b')

    def close(self):
        now = time.time()
        try:
            file.close(self)
            _md(self._destpath.dirname())
            self.finalpath = self._destpath
            os.rename(self.name, self.finalpath.path)
            os.utime(self.finalpath.path, (now, now))
        except:
            return defer.fail()
        return defer.succeed(self.finalpath)

    def abort(self):
        os.unlink(self.name)

_noItem = object()              # tag for optional argument to getItemByID
                                # default

class Store(Empowered):

    implements(iaxiom.IBeneficiary)

    transaction = None          # current transaction object

    storeID = -1                # I have a StoreID so that things can reference
                                # me

    store = property(lambda self: self) # I have a 'store' attribute because I
                                        # am 'stored' within myself; this is
                                        # also for references to use.

    def __init__(self, dbdir=None, debug=False, parent=None, idInParent=None):
        if parent is not None or idInParent is not None:
            assert parent is not None
            assert idInParent is not None
        self.parent = parent
        self.idInParent = idInParent
        self.debug = debug
        self.autocommit = True
        self.queryTimes = []
        self.execTimes = []
        self.statementCache = {} # non-normalized => normalized qmark SQL
                                 # statements

        self.objectCache = _fincache.FinalizingCache()


        if dbdir is None:
            dbfpath = IN_MEMORY_DATABASE
        else:
            dbdir = os.path.abspath(dbdir)
            dbfpath = os.path.join(dbdir, 'db.sqlite')
            self.filesdir = os.path.join(dbdir, 'files')
            if os.path.isdir(dbdir):
                if not os.path.exists(dbfpath):
                    raise OSError(
                        "The path %r is already a directory, "
                        "but not an Axiom Store" % (dbfpath,))
            else:
                _md(dbdir)
                _md(self.filesdir)
                _md(os.path.join(dbdir, 'temp'))
        self.dbdir = dbdir
        self.connection = sqlite.connect(dbfpath)
        self.cursor = self.connection.cursor()
        self.activeTables = {}  # tables which have had items added/removed
                                # this run

        # install powerups if we've never powered on before;
        create = not self.querySQL(_schema.HAS_SCHEMA_FEATURE,
                                   ['table', 'axiom_types'])[0][0]
        if create:
            for stmt in _schema.BASE_SCHEMA:
                self.executeSQL(stmt)

        # activate services if we have(?)
        # scheduler needs startup hook

        self.tableQueries = {}  # map typename: query string w/ storeID
                                # parameter.  a typename is a persistent
                                # database handle for what we'll call a 'FQPN',
                                # i.e. arg to namedAny.

        self.typenameToID = {} # map database-persistent typename to an oid in
                               # the types table

        self.typenameAndVersionToID = {} # obvious I hope

        self.idToTypename = {} # map database-persistent typeID (oid in types
                               # table) to typename

        self.service = None

        for oid, typename, version in self.querySQL(_schema.ALL_TYPES):
            self.typenameAndVersionToID[typename, version] = oid
            if version == _typeNameToMostRecentClass[typename].schemaVersion:
                self.typenameToID[typename] = oid
                self.idToTypename[oid] = typename
            else:
                self._prepareOldVersionOf(oid, typename, version)

        for typename in self.typenameToID:
            self.checkTypeSchemaConsistency(typename)


    def __repr__(self):
        return '<Store %s>' % (self.dbdir,)


    def newFile(self, *path):
        assert self.dbdir is not None, "Cannot create files in in-memory Stores (yet)"
        tmpname = os.path.join(self.dbdir, 'temp', str(tempCounter.next())+".tmp")
        return AtomicFile(
            tmpname,
            XFilePath(os.path.join(self.dbdir, 'files', *path)))

    def newDirectory(self, *path):
        assert self.dbdir is not None, "Cannot create directories in in-memory Stores (yet)"
        return FilePath(os.path.join(self.dbdir, 'files', *path))

    def checkTypeSchemaConsistency(self, typename):
        """
        Called for all known types at database startup: make sure that what we know
        (in memory) about this type is

        """
        # make sure that both the runtime and the database both know about this
        # type; if they don't both know, we can't check that their views are
        # consistent
        assert typename in self.typenameToID
        if typename not in _typeNameToMostRecentClass:
            print 'EARLY OUT CONSISTENCY CHECK: WHAT?'
            return
        typeID = self.typenameToID[typename]
        actualType = _typeNameToMostRecentClass[typename]
        #
        inMemorySchema = [(storedAttribute.indexed, storedAttribute.sqltype,
                           storedAttribute.allowNone, storedAttribute.columnName)
                          for (name, storedAttribute) in actualType.getSchema()]

        onDiskSchema = self.querySQL(_schema.IDENTIFYING_SCHEMA, [typeID])

        if inMemorySchema != onDiskSchema:
            raise RuntimeError(
                "Schema mismatch on already-loaded %r object version %d: %r != %r" %
                (actualType.typeName, actualType.schemaVersion, onDiskSchema, inMemorySchema))

        if self.querySQL(_schema.GET_TYPE_OF_VERSION,
                      [typename, actualType.schemaVersion]):
            raise RuntimeError(
                "Greater versions of database %r objects in the DB than in memory" %
                (typename,))

        # finally find old versions of the data and prepare to upgrade it.

    def _prepareOldVersionOf(self, typeID, typename, version):
        dummyItemSubclass(*self._dssargs(typeID, typename, version))

    def getOldVersionOf(self, typename, version):
        return _legacyTypes[typename, version]

    def _dssargs(self, typeID, typename, version):
        """
        Returns a 4-tuple suitable as args for dummyItemSubclass
        """

        appropriateSchema = self.querySQL(_schema.SCHEMA_FOR_TYPE, [typeID])
        # create actual attribute objects
        dummyAttributes = {}
        for indexed, pythontype, attribute, docstring in appropriateSchema:
            atr = getattr(attributes, pythontype)(indexed=indexed,
                                                  doc=docstring)
            dummyAttributes[attribute] = atr
        dummyBases = []
        retval = (typename, version, dummyAttributes, dummyBases)
        return retval


        # grab the schema for that version
        # look up upgraders which push it forward
        # insert "AutoUpgrader" class into idToTypename somehow(?)


    def query(self, tableClass,
              comparison=None,
              limit=None, offset=None,
              sort=None):
        if (tableClass.typeName,
            tableClass.schemaVersion) not in self.typenameAndVersionToID:
            return
        if comparison is not None:
            tables = set(comparison.getTableNames())
            where = ['WHERE', comparison.getQuery()]
            args = comparison.getArgsFor(self)
        else:
            tables = set()
            where = []
            args = []
        tables.add(tableClass.getTableName())
        query = ['SELECT',
                 tableClass.getTableName(), '.oid,',
                 tableClass.getTableName(), '.*', 'FROM',
                 ', '.join(tables)]
        query.extend(where)
        if sort is not None:
            query.append(sort)
        if limit is not None:
            # XXX LIMIT and OFFSET used to be using ?, but they started
            # generating syntax errors in places where generating the whole SQL
            # statement does not.  this smells like a bug in sqlite's parser to
            # me, but I don't know my SQL syntax standards well enough to be
            # sure -glyph
            query.append('LIMIT ')
            query.append(str(limit))
            if offset is not None:
                query.append('OFFSET ')
                query.append(str(offset))
        S = ' '.join(query)
        for row in self.querySQL(S, args):
            yield self._loadedItem(
                tableClass,
                row[0],
                row[1:])

    def _loadedItem(self, itemClass, storeID, attrs):
        if self.objectCache.has(storeID):
            result = self.objectCache.get(storeID)
            # XXX do checks on consistency between attrs and DB object, maybe?
        else:
            result = itemClass.existingInStore(self, storeID, attrs)
            if not result.__legacy__:
                self.objectCache.cache(storeID, result)
        return result

    def checkpoint(self):
        for item in self.transaction:
            # XXX: it should be possible here, using various clever hacks, to
            # automatically optimize functionally identical statements into
            # executemany.
            item.checkpoint()

    def revert(self):
        self.connection.rollback()
        for item in self.transaction:
            item.revert()

    def transact(self, f, *a, **k):
        if self.transaction is not None:
            return f(*a, **k)
        self.transaction = set()
        self.autocommit = False
        try:
            try:
                result = f(*a, **k)
                self.checkpoint()
            except:
                self.revert()
                raise
            else:
                self.commit()
                for committed in self.transaction:
                    committed.committed()
            return result
        finally:
            self.autocommit = True
            self.transaction = None

    def commit(self):
        if self.debug:
            print '*'*10, 'COMMIT', '*'*10
        self.connection.commit()

    def close(self):
        self.connection.close()
        if self.debug:
            if not self.queryTimes:
                print 'no queries'
            else:
                print 'query:', self.avgms(self.queryTimes)
            if not self.execTimes:
                print 'no execs'
            else:
                print 'exec:', self.avgms(self.execTimes)

    def avgms(self, l):
        return 'count: %d avg: %dus' % (len(l), int( (sum(l)/len(l)) * 1000000.),)


    def getTypeID(self, tableClass):
        """
        """
        key = (tableClass.typeName,
               tableClass.schemaVersion)
        if key in self.typenameAndVersionToID:
            return self.typenameAndVersionToID[key]

        sqlstr = []
        sqlarg = []
        indexes = []

        # needs to be calculated including version
        tableName = tableClass.getTableName()


        sqlstr.append("CREATE TABLE %s (" % tableName)

        for nam, atr in tableClass.getSchema():
            # it's a stored attribute
            sqlarg.append("\n%s %s" %
                          (atr.attrname, atr.sqltype))
            if atr.indexed:
                indexes.append(nam)

        sqlstr.append(', '.join(sqlarg))
        sqlstr.append(')')

        typeID = self.executeSQL(_schema.CREATE_TYPE, [tableClass.typeName,
                                                       tableClass.schemaVersion])

        for n, (name, storedAttribute) in enumerate(tableClass.getSchema()):
            self.executeSQL(
                _schema.ADD_SCHEMA_ATTRIBUTE,
                [typeID, n, storedAttribute.indexed, storedAttribute.sqltype,
                 storedAttribute.allowNone, storedAttribute.columnName,
                 storedAttribute.doc, storedAttribute.__class__.__name__])
            # XXX probably need something better for pythontype eventually,
            # when we figure out a good way to do user-defined attributes or we
            # start parameterizing references.

        self.typenameToID[tableClass.typeName] = typeID
        self.typenameAndVersionToID[key] = typeID
        self.idToTypename[typeID] = tableClass.typeName

        # The table and index creation code is executed last, because SQLite
        # has the extremely unfortunate habit of committing transactions as
        # soon as you issue a (CREATE|DROP) (TABLE|INDEX)

        # There is an obscure condition where the database could be left in an
        # inconsistent state.  If you do something that needs to be
        # transactional, create the first instance of a particular item type,
        # then do something which potentially raises an exception, then
        # checkpoint, THEN do something that depends on your first
        # transactional action, the database will commit as soon as the new
        # item type is created.

        # Note, however, that it is perfectly safe to do this _without_ a call
        # to checkpoint() because none of the Python-level SQL will run until
        # the database has checkpointed, either from user code or at the end of
        # the transaction.

        self.executeSQL(''.join(sqlstr))
        for index in indexes:
            self.executeSQL('CREATE INDEX axiomidx_%s_%s ON %s(%s)'
                            % (tableName, index,
                               tableName, index))

        return typeID

    def getTableQuery(self, typename, version):
        if typename not in self.tableQueries:
            query = 'SELECT * FROM %s WHERE oid = ?' % (
                TABLE_NAME(typename, version), )
            self.tableQueries[typename, version] = query
        return self.tableQueries[typename, version]

    def getItemByID(self, storeID, default=_noItem, autoUpgrade=True):
        """
        """
        assert storeID is not None
        if self.objectCache.has(storeID):
            return self.objectCache.get(storeID)
        results = self.querySQL(_schema.TYPEOF_QUERY, [storeID])
        assert (len(results) in [1, 0]),\
            "Database panic: more than one result for TYPEOF!"
        if results:
            typename, version = results[0]
            # for the moment we're going to assume no inheritance
            attrs = self.querySQL(self.getTableQuery(typename, version),
                                  [storeID])
            assert len(attrs) == 1, "No results for known-to-be-good object"
            attrs = attrs[0]
            useMostRecent = False
            moreRecentAvailable = False
            if typename in _typeNameToMostRecentClass:
                moreRecentAvailable = True
                mostRecent = _typeNameToMostRecentClass[typename]

                if mostRecent.schemaVersion < version:
                    raise RuntimeError("%s:%d - was found in the database and most recent %s is %d" %
                                       (typename, version, typename, mostRecent.schemaVersion))
                if mostRecent.schemaVersion == version:
                    useMostRecent = True
            if useMostRecent:
                T = mostRecent
            else:
                T = self.getOldVersionOf(typename, version)
            x = T.existingInStore(self, storeID, attrs)
            if moreRecentAvailable and (not useMostRecent) and autoUpgrade:
                x = upgrade.upgradeAllTheWay(x, typename, x.schemaVersion)
            if not x.__legacy__:
                # X is upgraded all the way
                self.objectCache.cache(storeID, x)
            return x
        if default is _noItem:
            raise KeyError(storeID)
        return default

    def _normalizeSQL(self, sql):
        assert "'" not in sql, "Strings are _NOT ALLOWED_"
        if sql not in self.statementCache:
            accum = []
            lines = sql.split('\n')
            for line in lines:
                line = line.split('--')[0]         # remove comments
                words = line.strip().split()
                accum.extend(words)
            normsql = ' '.join(accum)   # your SQL should never have any
                                        # significant whitespace in it, right?
            self.statementCache[sql] = normsql
        return self.statementCache[sql]


    def querySQL(self, sql, args=()):
        """For use with SELECT (or SELECT-like PRAGMA) statements.
        """
        sql = self._normalizeSQL(sql)
        if self.debug:
            result = timeinto(self.queryTimes, self._queryandfetch, sql, args)
        else:
            result = self._queryandfetch(sql, args)
        return result

    def _queryandfetch(self, sql, args):
        if self.debug:
            print '**', sql, '--', ', '.join(map(str, args))
        try:
            self.cursor.execute(sql, args)
        except (sqlite.ProgrammingError, sqlite.OperationalError), oe:
            raise RuntimeError("SQL: %r(%r) caused exception: %s" %(
                    sql, args, oe))
        result = self.cursor.fetchall()
        if self.autocommit:
            self.commit()
        if self.debug:
            print '  lastrow:', self.cursor.lastrowid
            print '  result:', result
        return result

    def executeSQL(self, sql, args=()):
        """
        For use with UPDATE, INSERT or CREATE statements.
        """
        sql = self._normalizeSQL(sql)
        if self.debug:
            rows = timeinto(self.execTimes, self._queryandfetch, sql, args)
        else:
            rows = self._queryandfetch(sql, args)
        assert not rows
        result = self.cursor.lastrowid
        return result


def timeinto(l, f, *a, **k):
    then = time.time()
    try:
        return f(*a, **k)
    finally:
        now = time.time()
        elapsed = now - then
        l.append(elapsed)

queryTimes = []
execTimes = []


class StorageService(MultiService):

    def __init__(self, *a, **k):
        MultiService.__init__(self)
        self.a = a
        self.k = k
        self.store = None

    def privilegedStartService(self):
        self.store = Store(*self.a, **self.k)
        IService(self.store).setServiceParent(self)
        MultiService.privilegedStartService(self)

    def close(self, ignored=None):
        # final close method, called after the service has been stopped
        self.store.close()
        self.store = None
        return ignored

    def stopService(self):
        return defer.maybeDeferred(
            MultiService.stopService, self).addBoth(
            self.close)
<commit_msg>"find or create" functionality Requested by Josh<commit_after># -*- test-case-name: axiom.test -*-

import time
import os
import itertools

from zope.interface import implements

from twisted.python.filepath import FilePath
from twisted.internet import defer
from twisted.python.reflect import qual
from twisted.application.service import IService, MultiService

from axiom import _schema, attributes, upgrade, _fincache, iaxiom

from pysqlite2 import dbapi2 as sqlite

from axiom.item import Item, \
    _typeNameToMostRecentClass, dummyItemSubclass,\
    _legacyTypes, TABLE_NAME, Empowered

IN_MEMORY_DATABASE = ':memory:'

tempCounter = itertools.count()

class XFilePath(FilePath):
    def dirname(self):
        return os.path.dirname(self.path)

def _md(dirname):
    if os.path.isdir(dirname):
        return False
    os.makedirs(dirname)
    return True

class AtomicFile(file):
    """I am a file which is moved from temporary to permanent storage when it
    is closed.

    After I'm closed, I will have a 'finalpath' property saying where I went.
    """

    implements(iaxiom.IAtomicFile)

    def __init__(self, tempname, destpath):
        self._destpath = destpath
        file.__init__(self, tempname, 'w+b')

    def close(self):
        now = time.time()
        try:
            file.close(self)
            _md(self._destpath.dirname())
            self.finalpath = self._destpath
            os.rename(self.name, self.finalpath.path)
            os.utime(self.finalpath.path, (now, now))
        except:
            return defer.fail()
        return defer.succeed(self.finalpath)

    def abort(self):
        os.unlink(self.name)

_noItem = object()              # tag for optional argument to getItemByID
                                # default

class Store(Empowered):

    implements(iaxiom.IBeneficiary)

    transaction = None          # current transaction object

    storeID = -1                # I have a StoreID so that things can reference
                                # me

    store = property(lambda self: self) # I have a 'store' attribute because I
                                        # am 'stored' within myself; this is
                                        # also for references to use.

    def __init__(self, dbdir=None, debug=False, parent=None, idInParent=None):
        if parent is not None or idInParent is not None:
            assert parent is not None
            assert idInParent is not None
        self.parent = parent
        self.idInParent = idInParent
        self.debug = debug
        self.autocommit = True
        self.queryTimes = []
        self.execTimes = []
        self.statementCache = {} # non-normalized => normalized qmark SQL
                                 # statements

        self.objectCache = _fincache.FinalizingCache()


        if dbdir is None:
            dbfpath = IN_MEMORY_DATABASE
        else:
            dbdir = os.path.abspath(dbdir)
            dbfpath = os.path.join(dbdir, 'db.sqlite')
            self.filesdir = os.path.join(dbdir, 'files')
            if os.path.isdir(dbdir):
                if not os.path.exists(dbfpath):
                    raise OSError(
                        "The path %r is already a directory, "
                        "but not an Axiom Store" % (dbfpath,))
            else:
                _md(dbdir)
                _md(self.filesdir)
                _md(os.path.join(dbdir, 'temp'))
        self.dbdir = dbdir
        self.connection = sqlite.connect(dbfpath)
        self.cursor = self.connection.cursor()
        self.activeTables = {}  # tables which have had items added/removed
                                # this run

        # install powerups if we've never powered on before;
        create = not self.querySQL(_schema.HAS_SCHEMA_FEATURE,
                                   ['table', 'axiom_types'])[0][0]
        if create:
            for stmt in _schema.BASE_SCHEMA:
                self.executeSQL(stmt)

        # activate services if we have(?)
        # scheduler needs startup hook

        self.tableQueries = {}  # map typename: query string w/ storeID
                                # parameter.  a typename is a persistent
                                # database handle for what we'll call a 'FQPN',
                                # i.e. arg to namedAny.

        self.typenameToID = {} # map database-persistent typename to an oid in
                               # the types table

        self.typenameAndVersionToID = {} # obvious I hope

        self.idToTypename = {} # map database-persistent typeID (oid in types
                               # table) to typename

        self.service = None

        for oid, typename, version in self.querySQL(_schema.ALL_TYPES):
            self.typenameAndVersionToID[typename, version] = oid
            if version == _typeNameToMostRecentClass[typename].schemaVersion:
                self.typenameToID[typename] = oid
                self.idToTypename[oid] = typename
            else:
                self._prepareOldVersionOf(oid, typename, version)

        for typename in self.typenameToID:
            self.checkTypeSchemaConsistency(typename)


    def __repr__(self):
        return '<Store %s>' % (self.dbdir,)


    def findOrCreate(self, Type, **attrs):

        andargs = []
        for k, v in attrs.iteritems():
            col = getattr(Type, k)
            andargs.append(col == v)

        for result in self.query(Type, attributes.AND(*andargs)):
            return result

        return Type(store=self, **attrs)


    def newFile(self, *path):
        assert self.dbdir is not None, "Cannot create files in in-memory Stores (yet)"
        tmpname = os.path.join(self.dbdir, 'temp', str(tempCounter.next())+".tmp")
        return AtomicFile(
            tmpname,
            XFilePath(os.path.join(self.dbdir, 'files', *path)))

    def newDirectory(self, *path):
        assert self.dbdir is not None, "Cannot create directories in in-memory Stores (yet)"
        return FilePath(os.path.join(self.dbdir, 'files', *path))

    def checkTypeSchemaConsistency(self, typename):
        """
        Called for all known types at database startup: make sure that what we know
        (in memory) about this type is

        """
        # make sure that both the runtime and the database both know about this
        # type; if they don't both know, we can't check that their views are
        # consistent
        assert typename in self.typenameToID
        if typename not in _typeNameToMostRecentClass:
            print 'EARLY OUT CONSISTENCY CHECK: WHAT?'
            return
        typeID = self.typenameToID[typename]
        actualType = _typeNameToMostRecentClass[typename]
        #
        inMemorySchema = [(storedAttribute.indexed, storedAttribute.sqltype,
                           storedAttribute.allowNone, storedAttribute.columnName)
                          for (name, storedAttribute) in actualType.getSchema()]

        onDiskSchema = self.querySQL(_schema.IDENTIFYING_SCHEMA, [typeID])

        if inMemorySchema != onDiskSchema:
            raise RuntimeError(
                "Schema mismatch on already-loaded %r object version %d: %r != %r" %
                (actualType.typeName, actualType.schemaVersion, onDiskSchema, inMemorySchema))

        if self.querySQL(_schema.GET_TYPE_OF_VERSION,
                      [typename, actualType.schemaVersion]):
            raise RuntimeError(
                "Greater versions of database %r objects in the DB than in memory" %
                (typename,))

        # finally find old versions of the data and prepare to upgrade it.

    def _prepareOldVersionOf(self, typeID, typename, version):
        dummyItemSubclass(*self._dssargs(typeID, typename, version))

    def getOldVersionOf(self, typename, version):
        return _legacyTypes[typename, version]

    def _dssargs(self, typeID, typename, version):
        """
        Returns a 4-tuple suitable as args for dummyItemSubclass
        """

        appropriateSchema = self.querySQL(_schema.SCHEMA_FOR_TYPE, [typeID])
        # create actual attribute objects
        dummyAttributes = {}
        for indexed, pythontype, attribute, docstring in appropriateSchema:
            atr = getattr(attributes, pythontype)(indexed=indexed,
                                                  doc=docstring)
            dummyAttributes[attribute] = atr
        dummyBases = []
        retval = (typename, version, dummyAttributes, dummyBases)
        return retval


        # grab the schema for that version
        # look up upgraders which push it forward
        # insert "AutoUpgrader" class into idToTypename somehow(?)


    def query(self, tableClass,
              comparison=None,
              limit=None, offset=None,
              sort=None):
        if (tableClass.typeName,
            tableClass.schemaVersion) not in self.typenameAndVersionToID:
            return
        if comparison is not None:
            tables = set(comparison.getTableNames())
            where = ['WHERE', comparison.getQuery()]
            args = comparison.getArgsFor(self)
        else:
            tables = set()
            where = []
            args = []
        tables.add(tableClass.getTableName())
        query = ['SELECT',
                 tableClass.getTableName(), '.oid,',
                 tableClass.getTableName(), '.*', 'FROM',
                 ', '.join(tables)]
        query.extend(where)
        if sort is not None:
            query.append(sort)
        if limit is not None:
            # XXX LIMIT and OFFSET used to be using ?, but they started
            # generating syntax errors in places where generating the whole SQL
            # statement does not.  this smells like a bug in sqlite's parser to
            # me, but I don't know my SQL syntax standards well enough to be
            # sure -glyph
            query.append('LIMIT ')
            query.append(str(limit))
            if offset is not None:
                query.append('OFFSET ')
                query.append(str(offset))
        S = ' '.join(query)
        for row in self.querySQL(S, args):
            yield self._loadedItem(
                tableClass,
                row[0],
                row[1:])

    def _loadedItem(self, itemClass, storeID, attrs):
        if self.objectCache.has(storeID):
            result = self.objectCache.get(storeID)
            # XXX do checks on consistency between attrs and DB object, maybe?
        else:
            result = itemClass.existingInStore(self, storeID, attrs)
            if not result.__legacy__:
                self.objectCache.cache(storeID, result)
        return result

    def checkpoint(self):
        for item in self.transaction:
            # XXX: it should be possible here, using various clever hacks, to
            # automatically optimize functionally identical statements into
            # executemany.
            item.checkpoint()

    def revert(self):
        self.connection.rollback()
        for item in self.transaction:
            item.revert()

    def transact(self, f, *a, **k):
        if self.transaction is not None:
            return f(*a, **k)
        self.transaction = set()
        self.autocommit = False
        try:
            try:
                result = f(*a, **k)
                self.checkpoint()
            except:
                self.revert()
                raise
            else:
                self.commit()
                for committed in self.transaction:
                    committed.committed()
            return result
        finally:
            self.autocommit = True
            self.transaction = None

    def commit(self):
        if self.debug:
            print '*'*10, 'COMMIT', '*'*10
        self.connection.commit()

    def close(self):
        self.connection.close()
        if self.debug:
            if not self.queryTimes:
                print 'no queries'
            else:
                print 'query:', self.avgms(self.queryTimes)
            if not self.execTimes:
                print 'no execs'
            else:
                print 'exec:', self.avgms(self.execTimes)

    def avgms(self, l):
        return 'count: %d avg: %dus' % (len(l), int( (sum(l)/len(l)) * 1000000.),)


    def getTypeID(self, tableClass):
        """
        """
        key = (tableClass.typeName,
               tableClass.schemaVersion)
        if key in self.typenameAndVersionToID:
            return self.typenameAndVersionToID[key]

        sqlstr = []
        sqlarg = []
        indexes = []

        # needs to be calculated including version
        tableName = tableClass.getTableName()


        sqlstr.append("CREATE TABLE %s (" % tableName)

        for nam, atr in tableClass.getSchema():
            # it's a stored attribute
            sqlarg.append("\n%s %s" %
                          (atr.attrname, atr.sqltype))
            if atr.indexed:
                indexes.append(nam)

        sqlstr.append(', '.join(sqlarg))
        sqlstr.append(')')

        typeID = self.executeSQL(_schema.CREATE_TYPE, [tableClass.typeName,
                                                       tableClass.schemaVersion])

        for n, (name, storedAttribute) in enumerate(tableClass.getSchema()):
            self.executeSQL(
                _schema.ADD_SCHEMA_ATTRIBUTE,
                [typeID, n, storedAttribute.indexed, storedAttribute.sqltype,
                 storedAttribute.allowNone, storedAttribute.columnName,
                 storedAttribute.doc, storedAttribute.__class__.__name__])
            # XXX probably need something better for pythontype eventually,
            # when we figure out a good way to do user-defined attributes or we
            # start parameterizing references.

        self.typenameToID[tableClass.typeName] = typeID
        self.typenameAndVersionToID[key] = typeID
        self.idToTypename[typeID] = tableClass.typeName

        # The table and index creation code is executed last, because SQLite
        # has the extremely unfortunate habit of committing transactions as
        # soon as you issue a (CREATE|DROP) (TABLE|INDEX)

        # There is an obscure condition where the database could be left in an
        # inconsistent state.  If you do something that needs to be
        # transactional, create the first instance of a particular item type,
        # then do something which potentially raises an exception, then
        # checkpoint, THEN do something that depends on your first
        # transactional action, the database will commit as soon as the new
        # item type is created.

        # Note, however, that it is perfectly safe to do this _without_ a call
        # to checkpoint() because none of the Python-level SQL will run until
        # the database has checkpointed, either from user code or at the end of
        # the transaction.

        self.executeSQL(''.join(sqlstr))
        for index in indexes:
            self.executeSQL('CREATE INDEX axiomidx_%s_%s ON %s(%s)'
                            % (tableName, index,
                               tableName, index))

        return typeID

    def getTableQuery(self, typename, version):
        if typename not in self.tableQueries:
            query = 'SELECT * FROM %s WHERE oid = ?' % (
                TABLE_NAME(typename, version), )
            self.tableQueries[typename, version] = query
        return self.tableQueries[typename, version]

    def getItemByID(self, storeID, default=_noItem, autoUpgrade=True):
        """
        """
        assert storeID is not None
        if self.objectCache.has(storeID):
            return self.objectCache.get(storeID)
        results = self.querySQL(_schema.TYPEOF_QUERY, [storeID])
        assert (len(results) in [1, 0]),\
            "Database panic: more than one result for TYPEOF!"
        if results:
            typename, version = results[0]
            # for the moment we're going to assume no inheritance
            attrs = self.querySQL(self.getTableQuery(typename, version),
                                  [storeID])
            assert len(attrs) == 1, "No results for known-to-be-good object"
            attrs = attrs[0]
            useMostRecent = False
            moreRecentAvailable = False
            if typename in _typeNameToMostRecentClass:
                moreRecentAvailable = True
                mostRecent = _typeNameToMostRecentClass[typename]

                if mostRecent.schemaVersion < version:
                    raise RuntimeError("%s:%d - was found in the database and most recent %s is %d" %
                                       (typename, version, typename, mostRecent.schemaVersion))
                if mostRecent.schemaVersion == version:
                    useMostRecent = True
            if useMostRecent:
                T = mostRecent
            else:
                T = self.getOldVersionOf(typename, version)
            x = T.existingInStore(self, storeID, attrs)
            if moreRecentAvailable and (not useMostRecent) and autoUpgrade:
                x = upgrade.upgradeAllTheWay(x, typename, x.schemaVersion)
            if not x.__legacy__:
                # X is upgraded all the way
                self.objectCache.cache(storeID, x)
            return x
        if default is _noItem:
            raise KeyError(storeID)
        return default

    def _normalizeSQL(self, sql):
        assert "'" not in sql, "Strings are _NOT ALLOWED_"
        if sql not in self.statementCache:
            accum = []
            lines = sql.split('\n')
            for line in lines:
                line = line.split('--')[0]         # remove comments
                words = line.strip().split()
                accum.extend(words)
            normsql = ' '.join(accum)   # your SQL should never have any
                                        # significant whitespace in it, right?
            self.statementCache[sql] = normsql
        return self.statementCache[sql]


    def querySQL(self, sql, args=()):
        """For use with SELECT (or SELECT-like PRAGMA) statements.
        """
        sql = self._normalizeSQL(sql)
        if self.debug:
            result = timeinto(self.queryTimes, self._queryandfetch, sql, args)
        else:
            result = self._queryandfetch(sql, args)
        return result

    def _queryandfetch(self, sql, args):
        if self.debug:
            print '**', sql, '--', ', '.join(map(str, args))
        try:
            self.cursor.execute(sql, args)
        except (sqlite.ProgrammingError, sqlite.OperationalError), oe:
            raise RuntimeError("SQL: %r(%r) caused exception: %s" %(
                    sql, args, oe))
        result = self.cursor.fetchall()
        if self.autocommit:
            self.commit()
        if self.debug:
            print '  lastrow:', self.cursor.lastrowid
            print '  result:', result
        return result

    def executeSQL(self, sql, args=()):
        """
        For use with UPDATE, INSERT or CREATE statements.
        """
        sql = self._normalizeSQL(sql)
        if self.debug:
            rows = timeinto(self.execTimes, self._queryandfetch, sql, args)
        else:
            rows = self._queryandfetch(sql, args)
        assert not rows
        result = self.cursor.lastrowid
        return result


def timeinto(l, f, *a, **k):
    then = time.time()
    try:
        return f(*a, **k)
    finally:
        now = time.time()
        elapsed = now - then
        l.append(elapsed)

queryTimes = []
execTimes = []


class StorageService(MultiService):

    def __init__(self, *a, **k):
        MultiService.__init__(self)
        self.a = a
        self.k = k
        self.store = None

    def privilegedStartService(self):
        self.store = Store(*self.a, **self.k)
        IService(self.store).setServiceParent(self)
        MultiService.privilegedStartService(self)

    def close(self, ignored=None):
        # final close method, called after the service has been stopped
        self.store.close()
        self.store = None
        return ignored

    def stopService(self):
        return defer.maybeDeferred(
            MultiService.stopService, self).addBoth(
            self.close)

{'fraction_non_alphanumeric': 0.086916492485319, 'fraction_numerical': 0.0012441524833283568, 'mean_word_length': 1.8185005961147345, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 42, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '862033', 'n_tokens_mistral': 10856, 'n_tokens_neox': 10383, 'n_words': 3113}