Catching and handling IntegrityError in SQLAlchemy

integrity_example.py

import sys

import sqlalchemy as sa
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

base = declarative_base()
print 'SQLAlchemy version {}'.format(sa.__version__)

USERS = [
  ('John', '[email protected]'),
  ('Bob', '[email protected]'),
  ('Jen', '[email protected]'),
  ('Jonathan', '[email protected]'),  # Second use of this address
  ('Sarah', '[email protected]')
]


class User(base):
  __tablename__ = 'user_integrity'
  id = sa.Column(sa.Integer, primary_key=True)
  email = sa.Column(sa.String(64), unique=True)
  name = sa.Column(sa.String(64))

  def __str__(self):
    return '<{cls.__name__}(email={self.email!r}, user={self.name!r})>'.format(
        cls=type(self), self=self)


def new_engine(echo=False):
  """Creates and returns a new database engine."""
  return sa.create_engine('mysql://play:play@/play', echo=echo)


def new_session(echo=True):
  """Creates and returns a new session object."""
  return sessionmaker(bind=new_engine(echo=echo))()


def create_schema():
  """Creates the database schema."""
  base.metadata.bind = new_engine()
  base.metadata.create_all()


def insert_user(session, name, email):
  """Adds a single User to the provided session, flushes and commits.

  If an IntegrityError is raised during flush, the transaction is rolled back
  and the User for the already existing email address is printed instead.
  """
  try:
    user = User(name=name, email=email)
    session.add(user)
    print '\nADDING {}'.format(user)
    session.flush()
  except IntegrityError:
    session.rollback()
    existing = session.query(User).filter_by(email=email).one()
    print '* INTEGRITY FAILURE, EMAIL IN USE: {}'.format(existing)
  else:
    session.commit()
    print '* SUCCESS'


def integrity_test():
  """Creates the table, adds a bunch of users with one duplicate email."""
  create_schema()
  session = new_session()
  for name, email in USERS:
    insert_user(session, name, email)


def purge_users():
  """Empties out the user table."""
  session = new_session()
  session.query(User).delete()
  session.commit()


def main():
  """Runs the SQLA IntegrityError demo or adds / deletes user records.

  If run without any commandline arguments, the script creates a table and adds
  a number of test users to it.

  If a name and email are provided, they are added to the table (if the table
  does not already exist, things break). Instead of providing a name and email,
  the single argument `PURGE` can be given, which deletes all records from the
  test user table.
  """
  if len(sys.argv) != 1:
    if sys.argv[1] == 'PURGE':
      return purge_users()
    return insert_user(new_session(), *sys.argv[1:])
  integrity_test()


if __name__ == '__main__':
  main()

output.log

(env)elmer@Penrose:~/devel$ python integrity.py 
<class 'sqlalchemy.event.base.DDLEventsDispatch'>
<class 'sqlalchemy.event.base.DDLEventsDispatch'>
SQLAlchemy version 0.9.7
<sqlalchemy.event.base.InstanceEventsDispatch object at 0x7fe27f1e9150>
<sqlalchemy.event.base.InstanceEventsDispatch object at 0x7fe27f1e9150>
<sqlalchemy.event.base.InstanceEventsDispatch object at 0x7fe27f1e9150>
<sqlalchemy.event.base.PoolEventsDispatch object at 0x7fe27f00f110>
<sqlalchemy.event.base.PoolEventsDispatch object at 0x7fe27f102dd0>

ADDING <User(email='[email protected]', user='John')>
2015-06-05 20:43:23,979 INFO sqlalchemy.engine.base.Engine SHOW VARIABLES LIKE 'sql_mode'
2015-06-05 20:43:23,979 INFO sqlalchemy.engine.base.Engine ()
2015-06-05 20:43:23,980 INFO sqlalchemy.engine.base.Engine SELECT DATABASE()
2015-06-05 20:43:23,981 INFO sqlalchemy.engine.base.Engine ()
2015-06-05 20:43:23,981 INFO sqlalchemy.engine.base.Engine show collation where `Charset` = 'utf8' and `Collation` = 'utf8_bin'
2015-06-05 20:43:23,981 INFO sqlalchemy.engine.base.Engine ()
2015-06-05 20:43:23,984 INFO sqlalchemy.engine.base.Engine SELECT CAST('test plain returns' AS CHAR(60)) AS anon_1
2015-06-05 20:43:23,984 INFO sqlalchemy.engine.base.Engine ()
2015-06-05 20:43:23,985 INFO sqlalchemy.engine.base.Engine SELECT CAST('test unicode returns' AS CHAR(60)) AS anon_1
2015-06-05 20:43:23,985 INFO sqlalchemy.engine.base.Engine ()
2015-06-05 20:43:23,986 INFO sqlalchemy.engine.base.Engine SELECT CAST('test collated returns' AS CHAR CHARACTER SET utf8) COLLATE utf8_bin AS anon_1
2015-06-05 20:43:23,986 INFO sqlalchemy.engine.base.Engine ()
2015-06-05 20:43:23,987 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2015-06-05 20:43:23,988 INFO sqlalchemy.engine.base.Engine INSERT INTO user_integrity (email, name) VALUES (%s, %s)
2015-06-05 20:43:23,988 INFO sqlalchemy.engine.base.Engine ('[email protected]', 'John')
2015-06-05 20:43:23,989 INFO sqlalchemy.engine.base.Engine COMMIT
* SUCCESS

ADDING <User(email='[email protected]', user='Bob')>
2015-06-05 20:43:23,991 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2015-06-05 20:43:23,991 INFO sqlalchemy.engine.base.Engine INSERT INTO user_integrity (email, name) VALUES (%s, %s)
2015-06-05 20:43:23,991 INFO sqlalchemy.engine.base.Engine ('[email protected]', 'Bob')
2015-06-05 20:43:23,992 INFO sqlalchemy.engine.base.Engine COMMIT
* SUCCESS

ADDING <User(email='[email protected]', user='Jen')>
2015-06-05 20:43:23,993 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2015-06-05 20:43:23,994 INFO sqlalchemy.engine.base.Engine INSERT INTO user_integrity (email, name) VALUES (%s, %s)
2015-06-05 20:43:23,994 INFO sqlalchemy.engine.base.Engine ('[email protected]', 'Jen')
2015-06-05 20:43:23,995 INFO sqlalchemy.engine.base.Engine COMMIT
* SUCCESS

ADDING <User(email='[email protected]', user='Jonathan')>
2015-06-05 20:43:23,995 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2015-06-05 20:43:23,996 INFO sqlalchemy.engine.base.Engine INSERT INTO user_integrity (email, name) VALUES (%s, %s)
2015-06-05 20:43:23,996 INFO sqlalchemy.engine.base.Engine ('[email protected]', 'Jonathan')
2015-06-05 20:43:23,996 INFO sqlalchemy.engine.base.Engine ROLLBACK
2015-06-05 20:43:23,999 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2015-06-05 20:43:24,000 INFO sqlalchemy.engine.base.Engine SELECT user_integrity.id AS user_integrity_id, user_integrity.email AS user_integrity_email, user_integrity.name AS user_integrity_name 
FROM user_integrity 
WHERE user_integrity.email = %s
2015-06-05 20:43:24,001 INFO sqlalchemy.engine.base.Engine ('[email protected]',)
* INTEGRITY FAILURE, EMAIL IN USE: <User(email='[email protected]', user='John')>

ADDING <User(email='[email protected]', user='Sarah')>
2015-06-05 20:43:24,002 INFO sqlalchemy.engine.base.Engine INSERT INTO user_integrity (email, name) VALUES (%s, %s)
2015-06-05 20:43:24,003 INFO sqlalchemy.engine.base.Engine ('[email protected]', 'Sarah')
2015-06-05 20:43:24,003 INFO sqlalchemy.engine.base.Engine COMMIT
* SUCCESS