else:
collist = string.join([c.fullname for c in inner_columns], ', ')
- text = "SELECT " + collist + " FROM "
+ text = "SELECT " + collist + " \nFROM "
whereclause = select.whereclause
if whereclause is not None:
t = self.get_str(whereclause)
if t:
- text += " WHERE " + t
+ text += " \nWHERE " + t
for tup in select._clauses:
text += " " + tup[0] + " " + self.get_str(tup[1])
self.context.transaction = None
self.context.tcount = None
- def execute(self, statement, parameters, connection = None, **params):
+ def execute(self, statement, parameters, connection = None, echo = None, **params):
if parameters is None:
parameters = {}
- if self._echo:
+ if echo is True or self._echo:
self.log(statement)
self.log(repr(parameters))
metadata = cursor.description
self.props = {}
i = 0
- for item in metadata:
- self.props[item[0]] = i
- self.props[i] = i
- i+=1
+ if metadata is not None:
+ for item in metadata:
+ self.props[item[0]] = i
+ self.props[i] = i
+ i+=1
def fetchone(self):
row = self.cursor.fetchone()
if row is not None:
- print repr(row)
+ #print repr(row)
return RowProxy(self, row)
else:
return None
return EagerLoader(mapper, secondary, primaryjoin, secondaryjoin, **options)
def relation_mapper(class_, selectable, secondary = None, primaryjoin = None, secondaryjoin = None, table = None, properties = None, lazy = True, **options):
- return relation_loader(mapper(class_, selectable, table = table, properties = properties, isroot = False), secondary, primaryjoin, secondaryjoin, lazy = lazy, **options)
+ return relation_loader(mapper(class_, selectable, table = table, properties = properties, isroot = False, **options), secondary, primaryjoin, secondaryjoin, lazy = lazy, **options)
-def mapper(class_, selectable, table = None, properties = None, identitymap = None, use_smart_properties = True, isroot = True):
- return Mapper(class_, selectable, table = table, properties = properties, identitymap = identitymap, use_smart_properties = use_smart_properties, isroot = isroot)
+def mapper(*args, **params):
+ return Mapper(*args, **params)
def identitymap():
return IdentityMap()
return LazyOption(name)
class Mapper(object):
- def __init__(self, class_, selectable, table = None, properties = None, identitymap = None, use_smart_properties = True, isroot = True):
+ def __init__(self, class_, selectable, table = None, properties = None, identitymap = None, use_smart_properties = True, isroot = True, echo = None):
self.class_ = class_
self.selectable = selectable
self.use_smart_properties = use_smart_properties
else:
self.table = table
self.props = {}
-
+
+ self.echo = echo
for column in self.selectable.columns:
self.props[column.key] = ColumnProperty(column)
def init(self, root):
self.root = root
self.identitymap = root.identitymap
+ self.echo = self.root.echo
[prop.init(key, self, root) for key, prop in self.props.iteritems()]
def instances(self, cursor):
return self._compile(whereclause, **options)
def options(self, *options):
+ """uses this mapper as a prototype for a new mapper with different behavior.
+
+ *options is a list of options directives, which include eagerload() and lazyload()"""
mapper = copy.copy(self)
for option in options:
option.process(mapper)
def _select_statement(self, statement, **params):
statement.use_labels = True
+ statement.echo = self.echo
return self.instances(statement.execute(**params))
def _identity_key(self, row):
def init(self, key, parent, root):
self.key = key
self.mapper.init(root)
+ if self.secondary is not None:
+ if self.secondaryjoin is None:
+ self.secondaryjoin = match_primaries(self.target, self.secondary)
+ if self.primaryjoin is None:
+ self.primaryjoin = match_primaries(parent.selectable, self.secondary)
+ else:
+ if self.primaryjoin is None:
+ self.primaryjoin = match_primaries(parent.selectable, self.target)
def save(self, object, traverse, refetch):
# if a mapping table does not exist, save a row for all objects
def init(self, key, parent, root):
PropertyLoader.init(self, key, parent, root)
if not hasattr(parent.class_, key):
+ if not issubclass(parent.class_, object):
+ raise "LazyLoader can only be used with new-style classes, i.e. subclass object"
setattr(parent.class_, key, SmartProperty(key).property())
-
+
def setup(self, key, primarytable, statement, **options):
- self.lazywhere = self.whereclause.copy_structure()
+ if self.secondaryjoin is not None:
+ self.lazywhere = sql.and_(self.primaryjoin, self.secondaryjoin)
+ else:
+ self.lazywhere = self.primaryjoin
+ self.lazywhere = self.lazywhere.copy_structure()
li = LazyIzer(primarytable)
self.lazywhere.accept_visitor(li)
self.binds = li.binds
class EagerLoader(PropertyLoader):
def init(self, key, parent, root):
PropertyLoader.init(self, key, parent, root)
-
- if self.secondary is not None:
- if self.secondaryjoin is None:
- self.secondaryjoin = match_primaries(self.target, self.secondary)
- if self.primaryjoin is None:
- self.primaryjoin = match_primaries(parent.selectable, self.secondary)
- else:
- if self.primaryjoin is None:
- self.primaryjoin = match_primaries(parent.selectable, self.target)
-
self.to_alias = util.Set()
[self.to_alias.append(f) for f in self.primaryjoin._get_from_objects()]
if self.secondaryjoin is not None:
statement.append_from(statement._outerjoin)
statement.append_column(self.target)
for key, value in self.mapper.props.iteritems():
- value.setup(key, self.mapper.selectable, statement)
+ value.setup(key, self.mapper.selectable, statement)
def execute(self, instance, row, identitykey, localmap, isduplicate):
"""receive a row. tell our mapper to look for a new object instance in the row, and attach
def process(self, mapper):
oldprop = mapper.props[self.key]
- mapper.set_property(self.key, EagerLoader(oldprop.mapper, oldprop.whereclause))
+ mapper.set_property(self.key, EagerLoader(oldprop.mapper, oldprop.secondary, primaryjoin = oldprop.primaryjoin, secondaryjoin = oldprop.secondaryjoin))
class LazyOption(MapperOption):
"""an option that switches a PropertyLoader to be a LazyLoader"""
return engine.compile(self, bindparams = bindparams)
def copy_structure(self):
+ """allows the copying of a statement's containers, so that a modified statement
+ can be produced without affecting the original. containing clauseelements,
+ like Select, Join, CompoundClause, BinaryClause, etc., should produce a copy of
+ themselves, whereas "leaf-node" clauseelements should return themselves."""
return self
def _engine(self):
# TODO: do pre-execute right here, for sequences, if the compiled object
# defines it
# TODO: why do we send the params twice, once to compile, once to c.get_params
- return e.execute(str(c), c.get_params(**params))
+ return e.execute(str(c), c.get_params(**params), echo = getattr(self, 'echo', None))
def result(self, **params):
e = self._engine()
c = property(lambda self: self.columns)
def accept_visitor(self, visitor):
+ print repr(self.__class__)
raise NotImplementedError()
def select(self, whereclauses = None, **params):
else:
self.label = self.column.name
self.fullname = self.column.name
+
+ def copy_structure(self):
+ return self.column
def _get_from_objects(self):
return [self.column.table]
memory = True
if memory:
- db = sqllite.engine(':memory:', {}, echo = True)
+ db = sqllite.engine(':memory:', {}, echo = False)
else:
if os.access('querytest.db', os.F_OK):
os.remove('querytest.db')
def setUp(self):
globalidentity().clear()
-
def testload(self):
"""tests loading rows with a mapper and producing object instances"""
def testoptions(self):
"""tests that a lazy relation can be upgraded to an eager relation via the options method"""
m = mapper(User, users, properties = dict(
- addresses = relation(Address, addresses, users.c.user_id==addresses.c.user_id, lazy = True)
- ))
+ addresses = relation(Address, addresses, lazy = True)
+ ), echo = True)
l = m.options(eagerload('addresses')).select()
print repr(l)
def testbasic(self):
"""tests a basic one-to-many lazy load"""
m = mapper(User, users, properties = dict(
- addresses = relation(Address, addresses, users.c.user_id==addresses.c.user_id, lazy = True)
- ))
+ addresses = relation(Address, addresses, lazy = True)
+ ), echo = True)
l = m.select(users.c.user_id == 7)
user = l[0]
a = user.addresses
items = orderitems
m = mapper(Item, items, properties = dict(
- keywords = relation(Keyword, keywords,
- and_(items.c.item_id == itemkeywords.c.item_id, keywords.c.keyword_id == itemkeywords.c.keyword_id), lazy = True),
- ))
- l = m.select()
- print repr(l)
+ keywords = relation(Keyword, keywords, itemkeywords, lazy = True),
+ ), echo = True)
+ # l = m.select()
+ # print repr(l)
l = m.select(and_(keywords.c.name == 'red', keywords.c.keyword_id == itemkeywords.c.keyword_id, items.c.item_id==itemkeywords.c.item_id))
- print repr(l)
-
-
-
+ print repr(l)
class EagerTest(PersistTest):
def testbasic(self):
"""tests a basic one-to-many eager load"""
m = mapper(User, users, properties = dict(
- addresses = relation(Address, addresses, lazy = False)
- ))
+ addresses = relation(Address, addresses, lazy = False),
+ ), echo = True)
l = m.select()
print repr(l)
criterion doesnt interfere with the eager load criterion."""
m = mapper(User, users, properties = dict(
addresses = relation(Address, addresses, primaryjoin = users.c.user_id==addresses.c.user_id, lazy = False)
- ))
+ ), echo = True)
l = m.select(and_(addresses.c.email_address == 'ed@lala.com', addresses.c.user_id==users.c.user_id))
print repr(l)
def testcompile(self):
"""tests deferred operation of a pre-compiled mapper statement"""
m = mapper(User, users, properties = dict(
- addresses = relation(Address, addresses, users.c.user_id==addresses.c.user_id, lazy = False)
+ addresses = relation(Address, addresses, lazy = False)
))
s = m.compile(and_(addresses.c.email_address == bindparam('emailad'), addresses.c.user_id==users.c.user_id))
c = s.compile()
def testmultieager(self):
"""tests eager loading with two relations simultaneously"""
m = mapper(User, users, properties = dict(
- addresses = relation(Address, addresses, users.c.user_id==addresses.c.user_id, lazy = False),
- orders = relation(Order, orders, users.c.user_id==orders.c.user_id, lazy = False),
+ addresses = relation(Address, addresses, primaryjoin = users.c.user_id==addresses.c.user_id, lazy = False),
+ orders = relation(Order, orders, lazy = False),
), identitymap = identitymap())
l = m.select()
print repr(l)
openorders = alias(orders, 'openorders')
closedorders = alias(orders, 'closedorders')
m = mapper(User, users, properties = dict(
- orders_open = relation(Order, openorders, and_(openorders.c.isopen == 1, users.c.user_id==openorders.c.user_id), lazy = False),
- orders_closed = relation(Order, closedorders, and_(closedorders.c.isopen == 0, users.c.user_id==closedorders.c.user_id), lazy = False)
+ orders_open = relation(Order, openorders, primaryjoin = and_(openorders.c.isopen == 1, users.c.user_id==openorders.c.user_id), lazy = False),
+ orders_closed = relation(Order, closedorders, primaryjoin = and_(closedorders.c.isopen == 0, users.c.user_id==closedorders.c.user_id), lazy = False)
), identitymap = identitymap())
l = m.select()
print repr(l)
"""tests eager loading, where one of the eager loaded items also eager loads its own
child items."""
ordermapper = mapper(Order, orders, properties = dict(
- items = relation(Item, orderitems, orders.c.order_id == orderitems.c.order_id, lazy = False)
+ items = relation(Item, orderitems, lazy = False)
))
m = mapper(User, users, properties = dict(
- addresses = relation(Address, addresses, users.c.user_id==addresses.c.user_id, lazy = False),
- orders = relation(ordermapper, users.c.user_id==orders.c.user_id, lazy = False),
+ addresses = relation(Address, addresses, lazy = False),
+ orders = relation(ordermapper, primaryjoin = users.c.user_id==orders.c.user_id, lazy = False),
))
l = m.select()
print repr(l)
m = mapper(Item, items, properties = dict(
keywords = relation(Keyword, keywords, itemkeywords, lazy = False),
- ))
+ ), echo = True)
l = m.select()
print repr(l)
l = m.select(and_(keywords.c.name == 'red', keywords.c.keyword_id == itemkeywords.c.keyword_id, items.c.item_id==itemkeywords.c.item_id))
print repr(l)
-
+
+ def testoneandmany(self):
+ items = orderitems
+
+ m = mapper(Item, items, properties = dict(
+ keywords = relation(Keyword, keywords, itemkeywords, lazy = False),
+ ))
+ m = mapper(Order, orders, properties = dict(
+ items = relation(m, lazy = False)
+ ), echo = True)
+ l = m.select("orders.order_id in (1,2,3)")
+ print repr(l)
+
+
+
if __name__ == "__main__":
unittest.main()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
