# relational algebra in Python 2.3

# Allows you to construct simple relational algebra queries in Python
# and translates them on the fly to SQL.  Then you can iterate over
# the query results, which are Python dictionaries.  I've tried this
# in Python 2.3, but maybe it might work in other versions of Python 2.

# This is just the barest of bare bones here, but I implemented it in
# one evening.  You can see the kind of stuff it supports by looking
# in the test() method down below.

# Credit goes to Avi Bryant's vaporware "Roe" for Smalltalk, to
# SchemeQL, and to E. F. Codd for inspiration.

# A short list of the most egregiously missing pieces:
# - union, intersect, difference
# - more joins than just the simple inner join
# - a more convenient simple inner join
# - aggregate operations (and thus 'group by')
# - other computations in the list of output columns
# - aliases for output columns
# - relops other than '=' in the where list
# - more convenient syntax for specifying foo.column('baz'),
#   e.g. foo['baz'] or foo.baz
# - support for query.column('foo.bar')
# - insertion and update of data

import types

class WideningProjection(Exception): pass
class AliasingAmbiguous(Exception): pass
class AmbiguousAttributeName(Exception): pass
class CantProjectColumnsFromAnotherTable(Exception): pass

class query:
        def __init__(self, tables, dbconn=None, where=[], columns=[]):
                (self.tables, self.where, self.columns, self.dbconn) = (
                        tables, where, columns, dbconn)
        def clonebut(self, **overrides):
                defaults = {
                        'tables': self.tables, 
                        'where': self.where,
                        'columns': self.columns,
                        'dbconn': self.dbconn,
                return self.__class__(**defaults)
        def sql(self):
                return 'select %(columns)s from %(tables)s%(where)s' % {
                        'columns': ', '.join(self.columnnames()),
                        'tables': self.tablespecs(),
                        'where': self.whereclause()
        def default_table(self):
                if len(self.tables) != 1: return None
                else: return self.tables[0][1]
        def columnnames(self):
                columns = self.columns or [
                        (alias, '*') for table, alias in self.tables]
                return [self.namefor(column) for column in columns]
        def tablespecs(self):
                rv = []
                for tablename, alias in self.tables:
                        if alias != tablename: 
                                rv.append('%s as %s' % (tablename, alias))
                        else: rv.append(tablename)
                return ', '.join(rv)
        def whereclause(self):
                if not self.where: return ''
                rv = []
                for variable, relop, value in self.where:
                        rv.append('%s %s %s' % (self.namefor(variable), relop, 
                return ' where ' + ' and '.join(rv)
        def project(self, *columns):
                ncolumns = []
                available_tables = [alias for tablename, alias in self.tables]
                for item in columns:
                        if type(item) in types.StringTypes:
                                item = self.column(item)
                        if item.tablename not in available_tables:
                                raise CantProjectColumnsFromAnotherTable(
                        ncolumns.append((item.tablename, item.attrname))
                if self.columns:
                        for col in ncolumns:
                                if col not in self.columns: 
                                        raise WideningProjection(col, self.columns)
                return self.clonebut(columns=ncolumns)
        def as(self, alias):
                # hmm, we need to handle renaming here somehow
                mytable = self.default_table()
                if not mytable: raise AliasingAmbiguous(alias, self.tables)
                return self.clonebut(tables=[(mytable, alias)])
        def __mul__(self, other):   # cartesian product
                # XXX shouldn't we be worrying about renaming
                # conflicts here?
                return self.clonebut(
                        tables=(self.tables + other.tables), 
                        where=(self.where + other.where),
                        columns=(self.columns + other.columns)
        def select(self, *conditions, **sugarclauses):
                eqclauses = [(self.column(var), '=', value) 
                        for (var, value) in sugarclauses.items()]
                return self.clonebut(where=self.where + eqclauses + list(conditions))
        def column(self, attrname):
                if not self.columns: return self.starattr(attrname)
                candidates = [(table, col) for (table, col) in self.columns
                        if col == attrname]
                if len(candidates) == 1:
                        return tablecolumn(self, candidates[0])
                        raise AmbiguousAttributeName(attrname, candidates)
        def starattr(self, attrname):
                # for when we're a "select *" query
                default = self.default_table()
                if not default:
                        raise AmbiguousAttributeName(attrname, 
                                [(tbl, attrname) for tbl in self.tables])
                return tablecolumn(self, (default, attrname))
        def namefor(self, thing):
                if type(thing) in types.StringTypes:
                        # quoting that works for Postgres and MySQL
                        # but breaks standard databases
                        return "'%s'" % thing.replace('\\', '\\\\').replace("'", "''")
                elif type(thing) is types.IntType:
                        return str(thing)
                elif type(thing) is types.TupleType:
                        tablename, attrname = thing
                elif hasattr(thing, 'tablename'):
                        tablename, attrname = thing.tablename, thing.attrname
                        raise TypeError(thing)
                if tablename == self.default_table(): return attrname
                else: return '%s.%s' % (tablename, attrname)
        def __iter__(self):
                return query_results(self.sql(), self.dbconn)

class tablecolumn:
        def __init__(self, query, (tablename, attrname)):
                (self.query, self.tablename, self.attrname) = (
                        query, tablename, attrname)
        def __eq__(self, other):
                return self, '=', other
        def name(self):
                return self.query.namefor(self)

class query_results:
        def __init__(self, sql, dbconn):
                self.results = dbconn.use_result()
        def __iter__(self): return self
        def next(self):
                rows = self.results.fetch_row(how=1)  # how=1 returns dicts
                if not rows: raise StopIteration
                return rows[0]

def table(tablename, dbconn=None):
        return query(tables=[(tablename, tablename)], dbconn=dbconn)

def ok(a, b): assert a == b, (a, b)

def test():
        foo = table("foo")
        ok(foo.sql(), 'select * from foo')
        ok(foo.project('a', 'b').sql(), 'select a, b from foo')
        ok(foo.as('bar').sql(), 'select * from foo as bar')
        ok((foo * foo.as('baz')).sql(), 
                'select foo.*, baz.* from foo, foo as baz')
        ok(foo.select(a=3).sql(), 'select * from foo where a = 3')
        ok(foo.select(a=5, b='asdf').sql(), "select * from foo where a = 5 and b = 
        ok(foo.select(b="Can't").sql(), "select * from foo where b = 'Can''t'")
        bar = table('bar')
        joinq = (foo * bar).select(bar.column('id') == 
foo.column('barid')).project(foo.column('a'), foo.column('b'), bar.column('d'))
        ok(joinq.sql(), "select foo.a, foo.b, bar.d from foo, bar where bar.id = 
        child = foo.as('child')
        ok((foo * child).select(foo.column('id') == child.column('parentid'))
                .sql(), "select foo.*, child.* from foo, foo as child "
                "where foo.id = child.parentid")


# I tested the MySQL connectivity as follows:
# import _mysql, relalg
# db = _mysql.connect(db='kragen')
# q = relalg.table('foo', db)
# list(q)
# list(q.select(a=3))
# list(q.select(a=4))

# Before this, I'd had to install the python-mysql Debian package to
# get the _mysql module, and I'd had to create the database, grant
# myself access, create a table, and put stuff in the table.  For
# future reference, MySQL doesn't have a 'varchar' type, just 'text',
# and a default MySQL installation allows "mysql -u root" to create
# databases and "grant all on newdatabasename.* to ''@'localhost'".
# Surprisingly, the '' around the @ really *are* important.

Reply via email to