def catch_db_error(f):
    """Catch and rollback database errors"""

    @functools.wraps(f)
    async def catching(self, *args, **kwargs):
        try:
            r = f(self, *args, **kwargs)
            if inspect.isawaitable(r):
                r = await r
        except SQLAlchemyError:
            self.log.exception("Rolling back session due to database error")
            self.db.rollback()
        else:
            return r

    return catching
Catch and rollback database errors
jupyterhub/utils.py