prototype SchemeQL/Roe-like thing in Python

Kragen Sitaker kragen at
Thu Apr 22 01:10:36 EDT 2004

# 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('')
# - 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, self.namefor(value)))
		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('bar').sql(), 'select * from foo as bar')
	ok((foo *'baz')).sql(), 
		'select foo.*, baz.* from foo, foo as baz')
	ok(, 'select * from foo where a = 3')
	ok(, b='asdf').sql(), "select * from foo where a = 5 and b = 'asdf'")
	ok("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 = foo.barid")
	child ='child')
	ok((foo * child).select(foo.column('id') == child.column('parentid'))
		.sql(), "select foo.*, child.* from foo, foo as child "
		"where = 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(
# list(

# 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.

More information about the Kragen-hacks mailing list