From: Mike Bayer Date: Sat, 29 Oct 2005 18:01:45 +0000 (+0000) Subject: (no commit message) X-Git-Tag: rel_0_1_0~420 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=2dff6b382c5659a350a636a755e52452c195c7dd;p=thirdparty%2Fsqlalchemy%2Fsqlalchemy.git --- diff --git a/TODO b/TODO index 4ed93f916f..ef4cebe7b4 100644 --- a/TODO +++ b/TODO @@ -1,32 +1,25 @@ TODO: +real exception classes instead of "raise 'string'" -mapper can take a row-processing/object instantiating function, without expensive construction - -lazy/eager loaders added to mapper via a list instaed of dictionary (do we ?) +lazy/eager loaders added to mapper via list arguments instead of dict? clean up select test suite -horizontal lazy/eager loaders for one object across tables - add it to existing lazy/eager loader -this will add a lot to SmartProperty since it needs to support a list of properties that all trigger -one lazy loader - -simple table inheritance +horizontal lazy loaders - this loads the data from an additional table +into the same object upon hitting any of its properties. -Oracle module +Oracle module - table reflection -Postgres module +Postgres module - table reflection MySQL module INSERT from a SELECT -extension types, like pickle type, user-defined types - generator functions returned from select() to handle [:] slices dynamically ? -make ResultProxy standard return value, get cursor accessible off of it and metadata stuff optional - -figure out why metadata doesnt work for sqllite row +examples -type bind param handling and result row handling +external functions able to be placed within a UOW's transaction commit; either at the head, or +based on dependencies to other mappers, individual objects diff --git a/lib/sqlalchemy/ansisql.py b/lib/sqlalchemy/ansisql.py index 22ea7e61d3..a551975401 100644 --- a/lib/sqlalchemy/ansisql.py +++ b/lib/sqlalchemy/ansisql.py @@ -201,9 +201,8 @@ class ANSICompiler(sql.Compiled): self.froms[select] = "(" + text + ")" - def visit_table(self, table): - self.froms[table] = table.name + self.froms[table] = table.fullname self.strings[table] = "" def visit_join(self, join): @@ -226,7 +225,7 @@ class ANSICompiler(sql.Compiled): self.binds[b.key] = b self.binds[b.shortname] = b - text = ("INSERT INTO " + insert_stmt.table.name + " (" + string.join([c[0].name for c in colparams], ', ') + ")" + + text = ("INSERT INTO " + insert_stmt.table.fullname + " (" + string.join([c[0].name for c in colparams], ', ') + ")" + " VALUES (" + string.join([self.bindparam_string(c[1].key) for c in colparams], ', ') + ")") self.strings[insert_stmt] = text @@ -245,7 +244,7 @@ class ANSICompiler(sql.Compiled): else: return self.get_str(p) - text = "UPDATE " + update_stmt.table.name + " SET " + string.join(["%s=%s" % (c[0].name, create_param(c[1])) for c in colparams], ', ') + text = "UPDATE " + update_stmt.table.fullname + " SET " + string.join(["%s=%s" % (c[0].name, create_param(c[1])) for c in colparams], ', ') if update_stmt.whereclause: text += " WHERE " + self.get_str(update_stmt.whereclause) @@ -253,7 +252,7 @@ class ANSICompiler(sql.Compiled): self.strings[update_stmt] = text def visit_delete(self, delete_stmt): - text = "DELETE FROM " + delete_stmt.table.name + text = "DELETE FROM " + delete_stmt.table.fullname if delete_stmt.whereclause: text += " WHERE " + self.get_str(delete_stmt.whereclause) @@ -270,7 +269,7 @@ class ANSISchemaGenerator(sqlalchemy.engine.SchemaIterator): raise NotImplementedError() def visit_table(self, table): - self.append("\nCREATE TABLE " + table.name + "(") + self.append("\nCREATE TABLE " + table.fullname + "(") separator = "\n" @@ -287,7 +286,7 @@ class ANSISchemaGenerator(sqlalchemy.engine.SchemaIterator): class ANSISchemaDropper(sqlalchemy.engine.SchemaIterator): def visit_table(self, table): - self.append("\nDROP TABLE " + table.name) + self.append("\nDROP TABLE " + table.fullname) self.execute() diff --git a/lib/sqlalchemy/schema.py b/lib/sqlalchemy/schema.py index bd0efc594b..17755676b0 100644 --- a/lib/sqlalchemy/schema.py +++ b/lib/sqlalchemy/schema.py @@ -61,7 +61,7 @@ class TableSingleton(type): except KeyError: if kwargs.get('mustexist', False): raise "Table '%s' not defined" % name - table = type.__call__(self, name, engine, *args) + table = type.__call__(self, name, engine, *args, **kwargs) engine.tables[name] = table # load column definitions from the database if 'autoload' is defined # we do it after the table is in the singleton dictionary to support @@ -76,7 +76,7 @@ class Table(SchemaItem): """represents a relational database table.""" __metaclass__ = TableSingleton - def __init__(self, name, engine, *args): + def __init__(self, name, engine, *args, **kwargs): self.name = name self.columns = OrderedProperties() self.c = self.columns @@ -85,6 +85,12 @@ class Table(SchemaItem): self.engine = engine self._impl = self.engine.tableimpl(self) self._init_items(*args) + self.schema = kwargs.get('schema', None) + if self.schema: + self.fullname = "%s.%s" % (self.schema, self.name) + else: + self.fullname = self.name + def reload_values(self, *args): self.columns = OrderedProperties() diff --git a/lib/sqlalchemy/sql.py b/lib/sqlalchemy/sql.py index e7f8a8ae17..700ae64fce 100644 --- a/lib/sqlalchemy/sql.py +++ b/lib/sqlalchemy/sql.py @@ -837,6 +837,14 @@ class UpdateBase(ClauseElement): if parameters is None: return None + if isinstance(parameters, list) or isinstance(parameters, tuple): + pp = {} + i = 0 + for c in self.table.c: + pp[c.key] = parameters[i] + i +=1 + parameters = pp + for key in parameters.keys(): value = parameters[key] if isinstance(value, Select): diff --git a/test/select.py b/test/select.py index 4db9447454..2227fc14ea 100644 --- a/test/select.py +++ b/test/select.py @@ -30,13 +30,23 @@ table3 = Table( Column('otherstuff', Integer), ) +table4 = Table( + 'remotetable', db, + Column('rem_id', Integer, primary_key=True), + Column('datatype_id', Integer), + Column('value', String(20)), + schema = 'remote_owner' +) + class SQLTest(PersistTest): - def runtest(self, clause, result, engine = None, params = None): + def runtest(self, clause, result, engine = None, params = None, checkparams = None): c = clause.compile(engine, params) self.echo("\n" + str(c) + repr(c.get_params())) cc = re.sub(r'\n', '', str(c)) self.assert_(cc == result, str(c) + "\n does not match \n" + result) - + if checkparams is not None: + self.assert_(c.get_params() == checkparams, "params dont match") + class SelectTest(SQLTest): def testtext(self): @@ -341,6 +351,13 @@ class CRUDTest(SQLTest): insert(table, dict(id = 3, name = 'jack')), "INSERT INTO mytable (myid, name) VALUES (:myid, :name)" ) + + # test with a tuple of params instead of named + self.runtest( + insert(table, (3, 'jack', 'mydescription')), + "INSERT INTO mytable (myid, name, description) VALUES (:myid, :name, :description)", + checkparams = {'myid':3, 'name':'jack', 'description':'mydescription'} + ) # insert with a subselect provided #self.runtest( @@ -374,6 +391,24 @@ class CRUDTest(SQLTest): def testdelete(self): self.runtest(delete(table, table.c.id == 7), "DELETE FROM mytable WHERE mytable.myid = :mytable_myid") +class SchemaTest(SQLTest): + def testselect(self): + self.runtest(table4.select(), "SELECT remotetable.rem_id, remotetable.datatype_id, remotetable.value FROM remote_owner.remotetable") + self.runtest(table4.select(and_(table4.c.datatype_id==7, table4.c.value=='hi')), "SELECT remotetable.rem_id, remotetable.datatype_id, remotetable.value FROM remote_owner.remotetable WHERE remotetable.datatype_id = :remotetable_datatype_id AND remotetable.value = :remotetable_value") + s = table4.select(and_(table4.c.datatype_id==7, table4.c.value=='hi')) + s.use_labels = True + self.runtest(s, "SELECT remotetable.rem_id AS remotetable_rem_id, remotetable.datatype_id AS remotetable_datatype_id, remotetable.value AS remotetable_value FROM remote_owner.remotetable WHERE remotetable.datatype_id = :remotetable_datatype_id AND remotetable.value = :remotetable_value") + + def testalias(self): + a = alias(table4, 'remtable') + self.runtest(a.select(a.c.datatype_id==7), "SELECT remtable.rem_id, remtable.datatype_id, remtable.value FROM remote_owner.remotetable remtable WHERE remtable.datatype_id = :remtable_datatype_id") + + def testupdate(self): + self.runtest(table4.update(table4.c.value=='test', values={table4.c.datatype_id:12}), "UPDATE remote_owner.remotetable SET datatype_id=:datatype_id WHERE remotetable.value = :remotetable_value") + + def testinsert(self): + self.runtest(table4.insert(values=(2, 5, 'test')), "INSERT INTO remote_owner.remotetable (rem_id, datatype_id, value) VALUES (:rem_id, :datatype_id, :value)") + if __name__ == "__main__": unittest.main()