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
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):
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
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)
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)
raise NotImplementedError()
def visit_table(self, table):
- self.append("\nCREATE TABLE " + table.name + "(")
+ self.append("\nCREATE TABLE " + table.fullname + "(")
separator = "\n"
class ANSISchemaDropper(sqlalchemy.engine.SchemaIterator):
def visit_table(self, table):
- self.append("\nDROP TABLE " + table.name)
+ self.append("\nDROP TABLE " + table.fullname)
self.execute()
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
"""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
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()
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):
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):
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(
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()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
