root/projects/sAsync/trunk/sasync/parray.py

# sAsync:
# An enhancement to the SQLAlchemy package that provides persistent
# dictionaries, text indexing and searching, and an access broker for
# conveniently managing database access, table setup, and
# transactions. Everything can be run in an asynchronous fashion using the
# Twisted framework and its deferred processing capabilities.
#
# Copyright (C) 2006-2007 by Edwin A. Suominen, http://www.eepatents.com
#
# This program is free software; you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation; either version 2 of the License, or (at your option) any later
# version.
# 
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE.  See the file COPYING for more details.
# 
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 51
# Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA

"""
Persistent Three-dimensional array objects
"""

# Imports
from twisted.internet import defer
import sqlalchemy as SA

from database import transact, AccessBroker
import search


NICENESS_WRITE = 6


class Transactor(AccessBroker):
    """
    I do the hands-on work of (potentially) non-blocking database access for
    the persistence of array elements within a uniquely-identified group.

    My methods return Twisted deferred instances to the results of their
    database accesses rather than forcing the client code to block while the
    database access is being completed.
    
    """
    def __init__(self, ID, *url, **kw):
        """
        Instantiates me for a three-dimensional array of elements within a
        particular group uniquely identified by the supplied integer I{ID},
        using a database connection to I{url}.
        """
        if not isinstance(ID, int):
            raise TypeError("Item IDs must be integers")
        self.groupID = ID
        if url:
            super(Transactor, self).__init__(url[0], **kw)
        else:
            super(Transactor, self).__init__()

    def startup(self):
        """
        You can run my transaction methods when the deferred returned from
        this method fires, and not before.
        """
        d = self.table(
            'sasync_array',
            SA.Column('group_id', SA.Integer),
            SA.Column('x', SA.Integer),
            SA.Column('y', SA.Integer),
            SA.Column('z', SA.Integer),
            SA.Column('value', SA.PickleType, nullable=False),
            unique_elements=['group_id', 'x', 'y', 'z']
            )
        return d
    
    @transact
    def load(self, x, y, z):
        """
        Element load transaction
        """
        array = self.sasync_array
        if not self.s('load'):
            self.s(
                [array.c.value],
                SA.and_(array.c.group_id == self.groupID,
                        array.c.x == SA.bindparam('x'),
                        array.c.y == SA.bindparam('y'),
                        array.c.z == SA.bindparam('z'))
                )
        rows = self.s().execute(x=hash(x), y=hash(y), z=hash(z)).fetchone()
        if not rows:
            return None
        else:
            return rows['value']

    @transact
    def update(self, x, y, z, value):
        """
        Element overwrite (entry update) transaction
        """
        elements = self.sasync_array
        u = elements.update(
            SA.and_(elements.c.group_id == self.groupID,
                    elements.c.x == hash(x),
                    elements.c.y == hash(y),
                    elements.c.z == hash(z))
            )
        u.execute(value=value)

    @transact
    def insert(self, x, y, z, value):
        """
        Element add (entry insert) transaction
        """
        self.sasync_array.insert().execute(
            group_id=self.groupID,
            x=hash(x), y=hash(y), z=hash(z), value=value)

    @transact
    def delete(self, x, y, z):
        """
        Element delete transaction
        """
        elements = self.sasync_array
        self.sasync_array.delete(
            SA.and_(elements.c.group_id == self.groupID,
                    elements.c.x == hash(x),
                    elements.c.y == hash(y),
                    elements.c.z == hash(z))
            ).execute()

    @transact
    def clear(self):
        """
        Transaction to clear all elements (B{Use with care!})
        """
        elements = self.sasync_array
        self.sasync_array.delete(
            elements.c.group_id == self.groupID).execute()


class PersistentArray(object):
    """
    I am a three-dimensional array of Python objects, addressable by any
    three-way combination of hashable Python objects. You can use me as a
    two-dimensional array by simply using some constant, e.g., C{None} when
    supplying an address for my third dimension.

    B{IMPORTANT}: Make sure you call my L{shutdown} method for an instance of
    me that you're done with before allowing that instance to be deleted.
    """
    search = None

    def __init__(self, ID, *url, **kw):
        """
        Constructor, with a URL and any engine-specifying keywords supplied if
        a particular engine is to be used for this instance. The following
        additional keyword is particular to this constructor:
        
        @keyword search: Set C{True} if text indexing is to be performed on items
            as they are written.

        """
        try:
            self.ID = hash(ID)
        except:
            raise TypeError("Item IDs must be hashable")
        if kw.pop('search', False):
            # No search object, worry about searching later
            self.search = None
        if url:
            self.t = Transactor(self.ID, url[0], **kw)
        else:
            self.t = Transactor(self.ID)
    
    def shutdown(self, *null):
        """
        Shuts down my database L{Transactor} and its synchronous task queue.
        """
        return self.t.shutdown()

    def write(self, funcName, *args, **kw):
        """
        Performs a database write transaction, returning a deferred to its
        completion.

        If we are updating the search index, there's a nuance to the
        deferred processing. In that case, when the write is done, the
        deferred is fired and processing separately proceeds with indexing
        of the written value. Here's how it works:

            1. Create a clean deferred B{d1} to return to the caller, whose
               callback(s) will be fired from the callback to the transaction's
               own deferred B{d2}.

            2. Start the write transaction and assign the C{writeDone} function
               as the callback to its deferred B{d2}. Note that the
               defer-to-queue transaction keeps a reference to the deferred
               object it instantiates, so we don't have to do so for either
               B{d2} or B{d3}. Those references are merely defined in the
               method for code readability.

        """
        def writeDone(noneResult, d1):
            x, y, z = [hash(arg) for arg in args[0:3]]
            document = "%d-%d" % (self.groupID, x)
            section = "%d-%d" % (y, z)
            d3 = self.search.index(
                value, document=document, section=section)
            d3.addCallback(self.search.ready)
            d1.callback(None)
        
        func = getattr(self.t, funcName)
        kwNew = {'niceness':kw['niceness']}
        if self.search is None:
            return func(*args, **kwNew)
        else:        
            d1 = defer.Deferred()
            self.search.busy()
            d2 = func(*args, **kwNew)
            d2.addCallback(writeDone, d1)
            return d1

    def get(self, x, y, z):
        """
        Retrieves an element (x,y,z) from the database.
        """
        d = self.t.dt.deferToAll()
        d.addCallback(lambda _: self.t.load(x, y, z))
        return d

    def set(self, x, y, z, value):
        """
        Persists the supplied I{value} of element (x,y,z) to the database,
        inserting or updating a row as appropriate.
        """
        def loaded(loadedValue):
            if loadedValue is None:
                return self.write(
                    "insert", x, y, z, value, niceness=NICENESS_WRITE)
            else:
                return self.write(
                    "update", x, y, z, value, niceness=NICENESS_WRITE)
        
        d = self.t.load(x, y, z)
        d.addCallback(loaded)
        self.t.dt.put(d)
        return d

    def delete(self, x, y, z):
        """
        Deletes the database row for element (x,y,z).
        """
        d = self.write("delete", x, y, z, niceness=NICENESS_WRITE)
        self.t.dt.put(d)
        return d

    def clear(self):
        """
        Deletes the entire group of database rows for U{all} of my elements
        (B{Use with care!})
        """
        d =self.write("clear", niceness=0)
        self.t.dt.put(d)
        return d


__all__ = ['PersistentArray']