Fellow SQLAlchemy users,
I have been using surrogate keys in all of my tables ever since I started using
an ORM (Hibernate, SQLObject, and now SQLAlchemy). I like surrogate keys
because they have no implicit meaning and they never change once they are
assigned. However, there are some cases (where a table would otherwise have a
multi-column primary key) where column(s) other than the PK must be unique. No
problem, add a UNIQUE constraint to the table definition. This has never been a
problem for me until recently.
Here's a very simplified version of my problem:
-- database table defs
create table orders (
id not null primary key,
...
)
create table items (
id not null primary key,
order_id int not null references orders,
alias varchar,
unique(alias, order_id)
)
# test code
item = Item(alias="test", order=order)
session.flush() # inserts item
order.items.remove(item)
item2 = Item(alias="test", order=order)
session.flush() # causes unique constraint violation
# because "item2" is inserted before "item" is deleted
So my question is: what should I do in this situation? Note: the delete and the
insert must happen within the same UnitOfWork (i.e. I can't do a flush()
between remove item and create item2).
My first idea was to look for an existing item with the same "unique" fields and "update"
that instead of doing both delete and insert. This involves tightly coupling my app with the internal
workings of UnitOfWork and/or HistoryArraySet (depending on whether session.delete() or list.remove() was
used to "delete" the item). I initially thought it might have worked if I had these:
Session.undelete(obj)
HistoryArraySet.undelete(obj)
Both of these methods would revert the deleted status of obj or do nothing if
the given object was not marked for deletion. But I eventually discarded the
idea entirely because HistoryArraySet was not reliable for adding/removing
items repetitively (for example, never use list.clear() because all history is
lost and there's no way to rollback). It gets especially hairy when an item is
in more than one managed collection at the same time.
It seems like the best way to solve this problem would be for SA to execute the
delete before the insert in the case where there would be a unique constraint
violation if the insert happened first. However, SA would need to know about
the unique constraint to do that.
Are there better ways to solve this problem? Are my table definitions flawed?
Thanks in advance for your suggestions.
~ Daniel
P.S. I have attached a unit test for this issue. It works on 0.2.
import testbase
from sqlalchemy import *
import string
import sqlalchemy.attributes as attr
class UniqueTest(testbase.AssertMixin):
def setUpAll(self):
self.install_threadlocal()
metadata = testbase.metadata
global line_items
line_items = Table("line_items", metadata,
Column("id", Integer, Sequence('line_items_id_seq', optional=True),
primary_key=True),
Column("name", String),
)
global item_details
item_details = Table("item_details", metadata,
Column("id", Integer, Sequence('item_details_id_seq',
optional=True), primary_key=True),
Column("line_item_id", Integer, ForeignKey("line_items.id")),
Column("alias", String),
)
line_items.create()
cn = testbase.db.connect()
cn.execute("""
CREATE TABLE item_details (
id SERIAL PRIMARY KEY,
line_item_id INT NOT NULL REFERENCES line_items,
alias TEXT NOT NULL,
UNIQUE (line_item_id, alias)
);""")
cn.close()
def tearDownAll(self):
item_details.drop()
line_items.drop()
#testbase.db.tables.clear()
self.uninstall_threadlocal()
def setUp(self):
objectstore.clear()
clear_mappers()
def tearDown(self):
item_details.delete().execute()
line_items.delete().execute()
def test_unique_constraint(self):
'''Test unique constraint with insert
'''
class LineItem(object):
def __init__(self, name=''):
self.name = name
class ItemDetail(object):
def __init__(self, alias='', line_item=None):
self.alias = alias
self.line_item = line_item
ItemDetail.mapper = mapper(ItemDetail, item_details)
LineItem.mapper = mapper(LineItem, line_items, properties=dict(
details=relation(ItemDetail, private=True, backref='line_item')
))
item = LineItem("test")
d1 = ItemDetail("a1", item)
self.assert_(d1 in item.details)
objectstore.flush()
item.details.remove(d1) # remove detail
d2 = ItemDetail("a1", item) # add another with same unique cols
(different PK)
self.assert_(d1 is not d2)
self.assert_(d2 in item.details)
objectstore.flush() # causes unique constraint violation
# because SA tries to insert d2 before deleting d1
if __name__ == "__main__":
testbase.main()
