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 [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.
--
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.
