Hi *,
I am fighting half a day with something I expected to be trivial: Keep
the order of items in a collection implemented vi a secondary table
(many-to-many relationship).
Basically, I have a Collection class with a relationship to Items in the
collection. That relationship is configured via
items=relation(Item,
secondary=collection_item_table,
order_by=[collection_item_table.c.item_order])
Now my problem is: How to update the item_order column in the table? So
far I did not find any way to do that. For a many-to-one relationship,
orderinglist will do this just fine.
I tried using a MapperExtension that goes through the relevant rows in
the secondary table and updates the item_order column (code attached).
It turns out that the after_insert and after_update extension points are
called before the child collections are flushed, so the primary keys of
any new items are not available at that time. Apart from that, it is a
bit heavy on the database to update all the item rows for a selection on
each change...
Another approach I tried was to use replace the relationship via custom
code querying the relation when reconstructing a collection instance.
The loading part works fine but I failed in populating the
collection_item_table in the first place, as the items must be flushed
before the Collection for the item_id to be available and I did not find
a way to tell SQLAlchemy of this dependency.
Any hint on how to do this is greatly appreciated.
Greetings, Torsten
--
DYNAmore Gesellschaft fuer Ingenieurdienstleistungen mbH
Torsten Landschoff
Office Dresden
Tel: +49-(0)351-4519587
Fax: +49-(0)351-4519561
mailto:torsten.landsch...@dynamore.de
http://www.dynamore.de
Registration court: Mannheim, HRB: 109659, based in Karlsruhe,
Managing director: Prof. Dr. K. Schweizerhof, Dipl.-Math. U. Franz
--
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to sqlalch...@googlegroups.com.
To unsubscribe from this group, send email to
sqlalchemy+unsubscr...@googlegroups.com.
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en.
#! /usr/bin/python
from sqlalchemy import *
from sqlalchemy.orm import *
# Set up the tables
meta = MetaData()
collection_table = Table("collection", meta,
Column("collection_id", Integer, primary_key=True))
item_table = Table("item", meta,
Column("item_id", Integer, primary_key=True),
Column("name", String))
collection_item_table = Table("collection_item", meta,
Column("collection_id", ForeignKey(collection_table.c.collection_id)),
Column("item_id", ForeignKey(item_table.c.item_id)),
Column("item_order", Integer))
# Mapped classes
class Collection(object):
def shallow_copy(self):
new = Collection()
new.items = self.items
return new
class Item(object):
def __init__(self, name):
self.name = name
# In a function to test without committing (which will work of course)
def maybe_commit(session):
session.commit()
pass
class CorrectOrderExtension(MapperExtension):
"""Updates the order of the entries in the collection_item_table to match with
the order in the items field of a Collection instance.
Does not work if items were not flushed before the Collection - how to force
flushing order? I would have expected that after_update/after_insert are called
after all collection attributes are completely written out."""
def after_update(self, mapper, connection, instance):
update = collection_item_table.update().where(
collection_item_table.c.collection_id==bindparam('b_collection')).where(
collection_item_table.c.item_id==bindparam('b_item_id')
).values(item_order=bindparam('b_item_order'))
collection_id = instance.collection_id
index = 0
updates = []
for item in instance.items:
item_id = item.item_id
assert item_id
updates.append(dict(
b_collection=collection_id,
b_item_id=item_id,
b_item_order=index))
index += 1
if updates:
connection.execute(update, updates)
return EXT_CONTINUE
def after_insert(self, mapper, connection, instance):
return self.after_update(mapper, connection, instance)
# Do the Object Relational Mapping
mapper(Item, item_table)
mapper(Collection, collection_table,
extension=CorrectOrderExtension(),
properties=dict(
items=relation(Item,
secondary=collection_item_table,
order_by=[collection_item_table.c.item_order])))
# Open database
engine = create_engine("sqlite:///", echo=True)
Session = sessionmaker(bind=engine)
meta.create_all(engine)
# Some items to play with
items = [Item(name) for name in ("foo", "bar", "baz", "qux")]
session = Session()
c = Collection()
c.items = items[:3]
session.add(c)
maybe_commit(session)
c2 = c.shallow_copy()
session.add(c2)
c2.items[1] = items[3]
maybe_commit(session)
# This is the expected final order...
assert [x.name for x in c2.items] == ["foo", "qux", "baz"]
#! /usr/bin/python
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.orm.interfaces import SessionExtension
meta = MetaData()
collection_table = Table("collection", meta,
Column("collection_id", Integer, primary_key=True))
item_table = Table("item", meta,
Column("item_id", Integer, primary_key=True),
Column("name", String))
collection_item_table = Table("collection_item", meta,
Column("collection_id", ForeignKey(collection_table.c.collection_id)),
Column("item_id", ForeignKey(item_table.c.item_id)),
Column("item_order", Integer))
class Collection(object):
def __init__(self):
self.items = []
def shallow_copy(self):
new = Collection()
new.items = self.items
return new
class Item(object):
def __init__(self, name):
self.name = name
def maybe_commit(session):
session.commit()
class ManageItemsExtension(MapperExtension):
"""Manual Implementation of a many to many relationship as a collection."""
def after_delete(self, mapper, connection, instance):
"""Remove secondary table entries when collection is deleted. Does not clean
orphaned items."""
collection_id = instance.collection_id
connection.execute(collection_item_table.delete().where(collection_id=collection_id))
return EXT_CONTINUE
def after_update(self, mapper, connection, instance):
"""Remove and recreate secondary table entries on update (room for optimization)."""
self.after_delete(mapper, connection, instance)
return self.after_insert(mapper, connection, instance)
def after_insert(self, mapper, connection, instance):
"""Add secondary table entries on inserting a collection."""
collection_id = instance.collection_id
items = instance.items
insert_op = collection_item_table.insert()
index = 0
inserts = []
for item in items:
item_id = item.item_id
assert item_id is not None
inserts.append(dict(collection_id=collection_id, item_id=item_id, item_order=index))
index += 1
if inserts:
connection.execute(insert_op, inserts)
return EXT_CONTINUE
def reconstruct_instance(self, mapper, instance):
"""Load the items via the secondary table when reconstructing from database."""
collection_id = instance.collection_id
instance.items = object_session(instance).query(Item).join(collection_item_table).filter(
collection_item_table.c.collection_id==collection_id).order_by(collection_item_table.c.item_order)
return EXT_CONTINUE
class AddItemsExtension(SessionExtension):
"""Attempt to ensure that items in a collection are added to the database and
flushed before the referencing Collection. Does NOT work!"""
def before_flush(self, session, flush_context, instances=None):
items = []
def process(instance):
if isinstance(instance, Collection):
for item in instance.items:
items.append(item)
for instance in session.dirty:
process(instance)
for instance in session.new:
process(instance)
if items:
session.add_all(items)
# session.flush() -> Exception, already in flush
mapper(Item, item_table)
mapper(Collection, collection_table, extension=ManageItemsExtension())
engine = create_engine("sqlite:///", echo=True)
Session = sessionmaker(bind=engine, extension=AddItemsExtension())
meta.create_all(engine)
session = Session()
items = [Item(name) for name in ("foo", "bar", "baz", "qux")]
session.add_all(items)
session.flush()
c = Collection()
c.items = items[:3]
session.add(c)
maybe_commit(session)
c2 = c.shallow_copy()
session.add(c2)
c2.items[1] = items[3]
maybe_commit(session)
c2_key = c2.collection_id
assert [x.name for x in c2.items] == ["foo", "qux", "baz"]
del session
print "-----------------"
session = Session()
c2 = session.query(Collection).get(c2_key)
assert [x.name for x in c2.items] == ["foo", "qux", "baz"]
