On 01/20/2012 10:42 AM, lestat wrote:
> Hi!
>
> Maybe anyone can tell how I can except this error?
>
>
> class TmpTest(db.Model, UnicodeMixin):
> __tablename__ = 'tmp_test'
>
> id = db.Column(db.Integer, primary_key=True)
> name = db.Column(db.String(50), nullable=False, unique=True)
> count = db.Column(db.Integer, nullable=False, default=0,
> server_default='0')
> time_create = db.Column(db.DateTime, nullable=False,
> default=func.now(), server_default=expression.text('now()'))
>
>
>
>
> from models import TmpTest
> from sqlalchemy.orm.exc import NoResultFound
> from app import db
> from multiprocessing import Pool, Process
>
> names = ['tmp_test_{0}'.format(num) for num in range(1000)]
>
> def sqlalchemy_test():
> for name in names:
> try:
> t = TmpTest.query.filter_by(name=name).one()
> except NoResultFound:
> t = TmpTest()
> t.name = name
> t.count = 0
> t.count += 1
> t = db.session.merge(t)
> db.session.add(t)
> db.session.commit()
>
>
> if __name__ == '__main__':
> p1 = Process(target=sqlalchemy_test)
> p1.daemon = True
> p2 = Process(target=sqlalchemy_test)
> p2.daemon = True
> p3 = Process(target=sqlalchemy_test)
> p3.daemon = True
> p4 = Process(target=sqlalchemy_test)
> p4.daemon = True
> p5 = Process(target=sqlalchemy_test)
> p5.daemon = True
> p6 = Process(target=sqlalchemy_test)
> p6.daemon = True
> p7 = Process(target=sqlalchemy_test)
> p7.daemon = True
> p8 = Process(target=sqlalchemy_test)
> p8.daemon = True
> p9 = Process(target=sqlalchemy_test)
> p9.daemon = True
> p10 = Process(target=sqlalchemy_test)
> p10.daemon = True
> p1.start()
> p2.start()
> p3.start()
> p4.start()
> p5.start()
> p6.start()
> p7.start()
> p8.start()
> p9.start()
> p10.start()
> p1.join()
> p2.join()
> p3.join()
> p4.join()
> p5.join()
> p6.join()
> p7.join()
> p8.join()
> p9.join()
> p10.join()
>
>
>
> IntegrityError: (IntegrityError) duplicate key value violates unique
> constraint "tmp_test_name_key"
> DETAIL: Key (name)=(tmp_test_45) already exists.
> 'INSERT INTO tmp_test (name, count, time_create) VALUES (%(name)s, %
> (count)s, now()) RETURNING tmp_test.id' {'count': 1, 'name':
> 'tmp_test_45'}
>
>
> IntegrityError: (IntegrityError) duplicate key value violates unique
> constraint "tmp_test_name_key"
> DETAIL: Key (name)=(tmp_test_26) already exists.
> 'INSERT INTO tmp_test (name, count, time_create) VALUES (%(name)s, %
> (count)s, now()) RETURNING tmp_test.id' {'count': 1, 'name':
> 'tmp_test_26'}
> Process Process-7:
>
>
> etc...
>
> Thanks!
>
>
AFAIK there is no standard way via SQLAlchemy/DBAPI to catch just a
particular constraint violation or even tell which constraint was
violated given an IntegrityError. The only way I have been able to do
this by extracting out the constraint name from the error text via
regular expressions. Unfortunately this technique is specific to both
your database vendor and possibly DBAPI implementation. For
PostgreSQL+psycopg2 I do something like:
try:
[...]
except IntegrityError as e:
if get_constraint_name(e) == "tmp_test_name_key":
print "Duplicate name"
else:
raise
def get_constraint_name(e):
# Unique constraint violations in PostgreSQL have error code 23505.
if e.orig.pgcode == "23505":
return re.search(r'^ERROR: duplicate key value violates unique
constraint "(.*?)"',
e.orig.pgerror).group(1)
else:
<handle other constraint types>
-Conor
--
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.
