Python sqlalchemy.orm.exc 模块，UnmappedInstanceError() 实例源码

def delete_good():
    """
    ??: ????
    ????: json
    ??: good_id(???ID)
    ????: json
    ??: status(1???, 0?json????, -1??????, -2?????), data(??????)
    """
    try:
        good = Good.query.get(request.json['good_id'])
        if not good:
            return jsonify({'status': -1, 'data': ['??????']})
        # db.session.delete(good.comments)
        db.session.delete(good)
        return jsonify({'status': 1, 'data': ["??comment"]})
    except KeyError as k:
        return jsonify({'status': 0, 'data': ['json????', k.args]})
    except UnmappedInstanceError as u:
        db.session.delete(good)
        return jsonify({'status': 1, 'data': ['?comment']})
    except Exception as e:
        return jsonify({'status': -2, 'data': ['????', e.args]})

def follow_post(id):
    user = session.query(User).filter_by(id=id).one()
    cuid = current_user.get_id()
    cuser = session.query(User).filter_by(id=cuid).one()
    try:
        if 'Unfollow' in request.form:
            session.add(cuser.unfollow(user))
            session.commit()
            flash("Good call unfollowing " + user.name +
                  ".  Nobody needs that.", "success")
            return redirect(url_for("entries"))
        elif 'Follow' in request.form:
            session.add(cuser.follow(user))
            session.commit()
            flash("You are now following " + user.name +
                  ".", "success")
            return redirect(url_for("stats_get"))
    except (IntegrityError, StaleDataError, UnmappedInstanceError):
        flash("That didn't work... mind letting Jon know?", "danger")
        session.rollback()
        return redirect(url_for("entries"))

def join_room(self, user_id: str, user_name: str, room_id: str, room_name: str) -> None:
        self.get_room_name(room_id)

        @with_session
        def _join_room(session=None):
            room = session.query(Rooms).filter(Rooms.uuid == room_id).first()
            user = session.query(Users).filter(Users.uuid == user_id).first()
            if user is None:
                user = Users()
                user.uuid = user_id
                user.name = user_name
                session.add(user)

            if room is None:
                logger.error('no such room %s (%s)' % (room_id, room_name))
                raise NoSuchRoomException(room_id)

            user.rooms.append(room)
            session.add(room)

            room.users.append(user)
            session.add(room)
            session.commit()

        try:
            _join_room()
        except UnmappedInstanceError as e:
            error_msg = 'user "%s" (%s) tried to join room "%s" (%s), but the room was None when joining; ' \
                        'likely removed after check and before joining: %s'
            logger.warning(error_msg % (user_id, user_name, room_id, room_name, str(e)))
        except IntegrityError as e:
            logger.warning('user "%s" (%s) tried to join room "%s" (%s) but it has been deleted: %s' %
                           (user_name, user_id, room_name, room_id, str(e)))
            raise NoSuchRoomException(room_id)

def get_bind(obj):
    """
    Return the bind for given SQLAlchemy Engine / Connection / declarative
    model object.

    :param obj: SQLAlchemy Engine / Connection / declarative model object

    ::

        from sqlalchemy_utils import get_bind


        get_bind(session)  # Connection object

        get_bind(user)

    """
    if hasattr(obj, 'bind'):
        conn = obj.bind
    else:
        try:
            conn = object_session(obj).bind
        except UnmappedInstanceError:
            conn = obj

    if not hasattr(conn, 'execute'):
        raise TypeError(
            'This method accepts only Session, Engine, Connection and '
            'declarative model objects.'
        )
    return conn

def save_inventory(station, downloaded_bytes_data):
    """Saves the inventory. Raises SqlAlchemyError if station's session is None,
    or (most likely IntegrityError) on failure
    """
    station.inventory_xml = dumps_inv(downloaded_bytes_data)
    try:
        object_session(station).commit()
    except UnmappedInstanceError:
        raise
    except SQLAlchemyError:
        object_session(station).rollback()
        raise