python - statement - Tratando de atrapar el error de integridad con SQLAlchemy

sqlalchemy table_args

Question

Estoy teniendo problemas al intentar detectar un error. Estoy usando Pyramid / SQLAlchemy e hice un formulario de registro con el correo electrónico como clave principal. El problema es que cuando se ingresa un correo electrónico duplicado, se genera un IntegrityError, por lo que estoy tratando de detectar ese error y enviar un mensaje, pero no importa lo que haga, no puedo detectarlo, el error sigue apareciendo.

try:
    new_user = Users(email, firstname, lastname, password)
    DBSession.add(new_user)
    return HTTPFound(location = request.route_url(''new''))
except IntegrityError:
    message1 = "Yikes! Your email already exists in our system. Did you forget your password?"

Recibo el mismo mensaje cuando lo intenté, except exc.SQLAlchemyError (aunque quiero detectar errores específicos y no una captura general). También probé exc.IntegrityError pero no exc.IntegrityError suerte (aunque existe en la API).

¿Hay algo mal con mi sintaxis de Python, o hay algo que necesito hacer especial en SQLAlchemy para atraparlo?

No sé cómo resolver este problema, pero tengo algunas ideas de lo que podría estar causando el problema. Tal vez la declaración de prueba no esté fallando, pero está teniendo éxito porque SQLAlchemy está generando la excepción y Pyramid está generando la vista, por lo que la except IntegrityError: nunca se activa. O, más probablemente, estoy captando este error completamente equivocado.

Answer 1

Así es como lo hago.

from contextlib import(
        contextmanager,
        )


@contextmanager
def session_scope():
    """Provide a transactional scope around a series of operations."""
    session = Session()
    try:
        yield session
        session.commit()
    except:
        session.rollback()
        raise
    finally:
        session.close()



def create_user(email, firstname, lastname, password):

    new_user = Users(email, firstname, lastname, password)

    try:

        with session_scope() as session:

            session.add(new_user)

    except sqlalchemy.exc.IntegrityError as e:
        pass

http://docs.sqlalchemy.org/en/latest/orm/session_basics.html#when-do-i-construct-a-session-when-do-i-commit-it-and-when-do-i-close-it

Answer 2

Edición: la respuesta editada arriba es una mejor manera de hacerlo, usando la reversión.

-

Si desea manejar transacciones en medio de una aplicación de pirámide o algo donde se realiza un compromiso de transacción automática al final de una secuencia, no hay magia que deba suceder.

Solo recuerde iniciar una nueva transacción si la transacción anterior ha fallado.

Me gusta esto:

def my_view(request):
   ... # Do things
   if success:
     try:
       instance = self._instance(**data)
       DBSession.add(instance)
       transaction.commit()
       return {''success'': True}
     except IntegrityError as e:  # <--- Oh no! Duplicate unique key
       transaction.abort()
       transaction.begin() # <--- Start new transaction
       return {''success'': False}

Observe que llamar a .commit () en una transacción exitosa está bien, por lo que no es necesario iniciar una nueva transacción después de una llamada exitosa.

Solo necesita abortar la transacción e iniciar una nueva si la transacción se encuentra en un estado fallido.

(Si la transacción no fuera así, podría usar un punto de salvaguarda y retroceder al punto de salvaguardia en lugar de iniciar una nueva transacción; lamentablemente, eso no es posible, ya que intentar una confirmación invalida un punto de salvaguarda anterior conocido. Muy bien, ¿no?) (Editar: <--- Resulta que estoy equivocado sobre eso ...)

Answer 3

En Pyramid, si ha configurado su sesión (lo que el andamio hace por usted automáticamente) para usar ZopeTransactionExtension , la sesión no se vacía ni se confirma hasta después de que se haya ejecutado la vista. Si desea capturar cualquier error de SQL en su opinión, debe forzar una flush para enviar el SQL al motor. DBSession.flush() debe hacerlo después de add(...) .

Actualizar

Estoy actualizando esta respuesta con un ejemplo de un punto de salvaguarda solo porque hay muy pocos ejemplos de cómo hacer esto con el paquete de transacción.

def create_unique_object(db, max_attempts=3):
    while True:
        sp = transaction.savepoint()
        try:
            obj = MyObject()
            obj.identifier = uuid.uuid4().hex
            db.add(obj)
            db.flush()
        except IntegrityError:
            sp.rollback()
            max_attempts -= 1
            if max_attempts < 1:
                raise
        else:
            return obj

obj = create_unique_object(DBSession)

Tenga en cuenta que incluso esto es susceptible de duplicarse entre transacciones si no se utiliza el bloqueo a nivel de tabla, pero al menos muestra cómo usar un punto de salvaguarda.

Answer 4

Lo que debe hacer es capturar una excepción general y generar su clase; Entonces puedes hacer la excepción más específica.

except Exception as ex:
    print ex.__class__

Answer 5

Puede que no haya operaciones de base de datos hasta que DBSession.commit() por lo tanto, IntegrityError se DBSession.commit() más adelante en la pila después de que el código del controlador que tiene try/except ya haya regresado.