On Jan 4, 2009, at 7:04 PM, Arn Vollebregt wrote:
>
> Hi,
>
> It is my understanding that when one adds an object instance to a
> session, SQLAlchemy checks for the same object based on it's primary
> key.
I wouldn't say that's accurate - it only looks at the existing
"identity" assigned to the given instance, based on what already-
loaded database state is present. If you put two brand new Foo()
objects in the session both with "id=1", SQLA wouldn't complain at
all. It lets the database do its job of detecting that constraint,
and the error is an IntegrityError raised by the DBAPI within the flush.
> Does
> this mean SQLAlchemy has no knowledge of the unique constraints in the
> table definition?
that is correct. The "unique" flag is only used for DDL generation,
or if the application wants to inspect the Table itself to make
decisions.
> And if so, is there a way to change this behavior,
> without querying the database?
SQLA core never chooses to reinvent functionality that's the database
already provides, so you'd have to build something on your end which
hashes all instances in a dictionary based on the desired unique
attributes. I've hopefully made this easy via the attached example
(SQLA 0.5, py2.5). If it works for you we should add it to the wiki's
UsageRecipes section.
--~--~---------~--~----~------------~-------~--~----~
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To post to this group, send email to [email protected]
To unsubscribe from this group, send email to
[email protected]
For more options, visit this group at
http://groups.google.com/group/sqlalchemy?hl=en
-~----------~----~----~----~------~----~------~--~---
from sqlalchemy.orm import object_session, column_property
from sqlalchemy.orm.interfaces import SessionExtension, AttributeExtension
import weakref
def _unique_dict(session):
"""Returns a weak valued dictionary associated with a Session which tracks constrained values.
Keys are tuples of the form:
(cls, (constraint_key1, constraint_key2, ...), (value1, value2, ...))
Values are instances of cls which are the current holders of the constraint
value indicated by the key.
"""
if not hasattr(session, '_unique_dict'):
session._unique_dict = weakref.WeakValueDictionary()
return session._unique_dict
class UniqueConstraintChecker(SessionExtension):
"""Listens for attach events and checks against a constraint dictionary for violations.
The dictionary is of the form {cls:(attr1, attr2, ...), cls:(attr1, attr2, ...), ...} indicating
classes and a tuple of attribute names which comprise a unique constraint.
"""
def __init__(self, constraints):
self._constraints = constraints
def after_attach(self, session, instance):
cls = instance.__class__
if cls in self._constraints:
unique_dict = _unique_dict(session)
constraint = self._constraints[cls]
composed_value = tuple([getattr(instance, k, None) for k in constraint])
if (cls, constraint, composed_value) in unique_dict:
raise Exception("Unique key %r already present" % (composed_value, ))
unique_dict[(cls, constraint, composed_value)] = instance
class UniqueConstraintListener(AttributeExtension):
"""Listens for set events on a single attribute and checks for constraint violations
against a composed constraint definition.
"""
def __init__(self, constraint, attrname):
self.constraint = constraint
self.attrname = attrname
def set(self, state, value, oldvalue, initiator):
instance = state.obj()
session = object_session(instance)
if session:
cls = instance.__class__
unique_dict = _unique_dict(session)
old_composed_value = tuple([oldvalue if k == self.attrname else getattr(instance, k, None) for k in self.constraint])
new_composed_value = tuple([value if k == self.attrname else getattr(instance, k, None) for k in self.constraint])
if (cls, self.constraint, new_composed_value) in unique_dict:
raise Exception("Unique key %r already present" % (new_composed_value, ))
else:
del unique_dict[(cls, self.constraint, old_composed_value)]
unique_dict[(cls, self.constraint, new_composed_value)] = instance
return value
def UniqueColumn(*args, **kw):
"""Produce a declarative column_property() with a unique constraint checker."""
constraint = kw.pop('constraint')
attrname = kw.pop('attrname')
return column_property(
Column(*args, **kw),
extension=UniqueConstraintListener(constraint, attrname)
)
if __name__ == '__main__':
from sqlalchemy import Column, Integer, create_engine, UniqueConstraint
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
engine = create_engine('sqlite://', echo=True)
Base = declarative_base()
user_constraint = ('foo', 'bar')
class User(Base):
__tablename__ = 'user'
id = Column(Integer, primary_key=True)
foo = UniqueColumn(Integer, nullable=False, attrname='foo', constraint=user_constraint)
bar = UniqueColumn(Integer, nullable=False, attrname='bar', constraint=user_constraint)
__table_args__ = (UniqueConstraint(*user_constraint),{})
Base.metadata.create_all(engine)
sess = sessionmaker(
bind=engine,
extension=UniqueConstraintChecker({User:user_constraint})
)()
u1 = User(foo=5, bar=12)
u2 = User(foo=6, bar=12)
u3 = User(foo=6, bar=9)
sess.add_all([u1, u2, u3])
try:
sess.add(User(foo=6, bar=12)) # can't add a new object with a duplicate
except Exception, e:
print e
else:
assert False
try:
u2.foo = 5 # can't change an object already in the session to a duplicate key
except Exception, e:
print e
else:
assert False
# change the original to something else, now you can change u2.foo to 5
u1.bar = 19
u2.foo = 5
sess.commit()
