if you want to rethrow DBAPI exceptions you'd need to catch it at some point. We don't have a hook where you can build in a "try:/except:" around executions right now so such a thing would need to be called externally to SQLAlchemy:
def execute(connection, stmt): try: return connection.execute(stmt) except IntegrityError, err: if err.orig.pgcode == '23505': raise DuplicateKeyError() else: raise which doesn't really get at session.commit(). There's ways to monkeypatch such a function into the Connection itself, though that's kind of brittle. Since catching this error is something you want to do occasionally I'd likely just build a context manager: class catch_dupes(object): def __enter__(self): return None def __exit__(self, type, value, traceback): if issubclass(type, IntegrityError) and \ value.orig.pgcode == "23505": raise DuplicateKeyError() try: with catch_dupes(): db.session.commit() except DuplicateKeyError: # ... pretty ugly. if anyone else has some nifty python tricks here feel free. On Mar 30, 2012, at 7:28 AM, lestat wrote: > Hi! > > I use postgresql and I try create my exception DuplicateKeyError inheritanced > from IntegrityError. > > I know that duplicate key in postgresql has code 23505 and I must catch with > if err.orig.pgcode == '23505': > ... > > How I can create this? > > I don't work with many exceptions, and don't find what method it execute when > it raised. > > I try: > > from sqlalchemy.exc import IntegrityError > > > class DuplicateKeyError(IntegrityError): > def __init__(self, *args, **kwargs): > print '______________DuplicateKeyError init called' > print '____ args:',args > print '____ kwargs:',kwargs > super(DuplicateKeyError, self).__init__(*args, **kwargs) > > @classmethod > def instance(cls, *args, **kwargs): > print '______________DuplicateKeyError instance called' > print '____ cls:',cls > print '____ args:',args > print '____ kwargs:',kwargs > super(DuplicateKeyError, cls).instance(*args, **kwargs) > > @staticmethod > def __new__(cls, *args, **kwargs): > print '______________DuplicateKeyError new called' > print '____ cls:',cls > print '____ args:',args > print '____ kwargs:',kwargs > super(DuplicateKeyError, cls).__new__(*args, **kwargs) > > def __call__(self, *args, **kwargs): > print '______________DuplicateKeyError call called' > print '____ self:',self > print '____ args:',args > print '____ kwargs:',kwargs > super(DuplicateKeyError, self).__call__(*args, **kwargs) > > def test(): > try: > i = Interest(name='x1') > db.session.add(i) > db.session.commit() > except DuplicateKeyError as err: > print '===== err:',err > print '===== err.orig.pgcode:',err.orig.pgcode > > > but nothing of my prints does not executed. > > Where I wrong? > > Thanks! > > > -- > You received this message because you are subscribed to the Google Groups > "sqlalchemy" group. > To view this discussion on the web visit > https://groups.google.com/d/msg/sqlalchemy/-/ExQsD7uUi6MJ. > To post to this group, send email to sqlalchemy@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. -- You received this message because you are subscribed to the Google Groups "sqlalchemy" group. To post to this group, send email to sqlalchemy@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.