Python models 模块，User() 实例源码

def registration():
    if g.user is not None and g.user.is_authenticated():
        flash('Please Log out before registration')
        return redirect(url_for('index'))
    form = RegistrationForm()
    if request.method == 'POST' and form.validate_on_submit():
        """ check is it avaliable login """
        if User.query.filter_by(login=form.login.data).first():
            """ Create error message """
            flash('Choose another login', 'error')
        else:
            """ Add new user to DB """
            new_user = User(login=form.login.data, 
                password=bcrypt.generate_password_hash(form.password.data),
                name=form.name.data,
                surname=form.surname.data)
            db.session.add(new_user)
            db.session.commit()
            """ Success message """
            flash('Done')
            return redirect(url_for("login"))
    return render_template('registration.html',
        title='Registration',
        form=form)

def add_user():
    content = request.get_json(silent=True)
    user = User(**content.get('user'))
    try:
        db.query("INSERT INTO users " +
                 "(email, password) " +
                 "VALUES ('{0}', '{1}')".format(user.email,
                                               user.password))
    except IntegrityError as e:
        logger.info("User exists for {0}".format(user.email))
        return jsonify({"error": "Email {0} already has an user".format(user.email)})

    rows = db.query("SELECT seq FROM sqlite_sequence WHERE NAME = 'users'")
    user_id = rows.all()[0].get('seq', 0)

    created = db.query("SELECT * FROM users WHERE id = {0}".format(user_id))
    return jsonify(**created.all()[0])

def oauth_callback(provider):
    if not current_user.is_anonymous:
        return redirect(url_for('index'))
    oauth = OAuthSignIn.get_provider(provider)
    xd_id, name, email = oauth.callback()
    if xd_id is None:
        flash('Authentication failed.')
        return redirect(url_for('index'))
    user = User.query.filter_by(xd_id=xd_id).first()
    if not user: # Create, add and login new user. Redirect to /register
        user = User(xd_id=xd_id, name=name, email=email)
        db.session.add(user)
        db.session.commit()
        login_user(user, True)
        return redirect(url_for('dash')) # previously register
    else: # Login new user. Redirect to /
        login_user(user, True)
        return redirect(url_for('index'))

def update_user_details(user_id, name, email, new_password):
    """
    Update an existing user, identified by user_id, with the provided fields. If name or email is None, this will
    remove these fields from the user entry. If new_password is None, the user's password will not be updated.

    :param user_id: User ID of the user to update
    :param name: Updated name, can be empty string or None to indicate no update
    :param email: Updated email, can be empty string or None to indicate no update
    :param new_password: New password, if updating the user's password
    :return: models.User object representing the updated user
    :raises InvalidEmailException: If an invalid email is passed
    """
    if email and not is_email_address_valid(email):
        raise InvalidEmailException('{email_addr} is not a valid email address'.format(email_addr=email))

    user = get_user_by_id(user_id, active_only=True)
    user.name = name
    user.email = email
    if new_password:
        user.password_hash = util.cryptography.secure_hash(new_password)
    session.commit()
    return user

def get_user_by_id(user_id, active_only=False):
    """
    Get a User object by user_id, whose attributes match those in the database.

    :param user_id: User ID to query by
    :param active_only: Set this flag to True to only query for active users
    :return: User object for that user ID
    :raises UserDoesNotExistException: If no user exists with the given user_id
    """
    if active_only:
        user = models.User.query.filter_by(user_id=user_id, is_active=True).first()
    else:
        user = models.User.query.filter_by(user_id=user_id).first()
    if not user:
        raise UserDoesNotExistException('No user with user_id {user_id} exists'.format(user_id=user_id))
    return user

def get_user_by_username(username, active_only=False):
    """
    Get a User object by username, whose attributes match those in the database.

    :param username: Username to query by
    :param active_only: Set this flag to True to only query for active users
    :return: User object for that username
    :raises UserDoesNotExistException: If no user exists with the given username
    """
    if active_only:
        user = models.User.query.filter_by(username=username.lower(), is_active=True).first()
    else:
        user = models.User.query.filter_by(username=username.lower()).first()
    if not user:
        raise UserDoesNotExistException('No user with username {username} exists'.format(username=username))
    return user

def get_user_by_api_key(api_key, active_only=False):
    """
    Get a User object by api_key, whose attributes match those in the database.

    :param api_key: API key to query by
    :param active_only: Set this flag to True to only query for active users
    :return: User object for that user ID
    :raises UserDoesNotExistException: If no user exists with the given user_id
    """
    if active_only:
        user = models.User.query.filter_by(api_key=api_key, is_active=True).first()
    else:
        user = models.User.query.filter_by(api_key=api_key).first()
    if not user:
        raise UserDoesNotExistException('No user with api_key {api_key} exists'.format(api_key=api_key))
    return user

def create_demo_user():
    #check to see if demo user already exists in the db
    if models.db_session.query(models.User).filter(models.User.profile_id=='demo_id').first() is None:
        #Create demo user and all demo user's associated relatives
        demo_user = models.User('demo_id', None, 'Lilly', 'Demo', None, None, None, None, demo.genome_data)
        for relative in demo.relatives:
            #Create a new relative with the information being passed from relatives_response
            new_relative = models.Relative(None, relative['first_name'], relative['last_name'], relative['sex'], relative['residence'], relative['similarity'], relative['maternal_side'], relative['paternal_side'], relative['picture_url'], relative['birth_year'], relative['relationship'], relative['birthplace'], relative['ancestry'])
            # Appending each relative to the demo user's relative property
            demo_user.relatives.append(new_relative)
            models.db_session.add(new_relative)
        # Add the demo user to the database and commit it
        models.db_session.add(demo_user)
        models.db_session.commit()


#CreateNewUser will be called in server.py when a user logging in has not been found in database

def createNewUser(name_response, relatives_response, genotype_response, user_response):
    #Grab the dnaPairs at relative snps
    genome_data = genotype_response.json().pop()
    #Define the user's basic information
    user_first_name = name_response.json()['first_name']
    user_last_name = name_response.json()['last_name']
    user_id = genome_data['id']
    user_email = user_response.json()
    #Create a new user following the Users Model
    new_user = models.User(user_id, user_email['email'], user_first_name, user_last_name, None, None, None, None, genome_data)
    #iterate through list of relatives
    for relative in relatives_response.json()['relatives']:
        #Create a new relative with the information being passed from relatives_response
        new_relative = models.Relative(None, relative['first_name'], relative['last_name'], relative['sex'], relative['residence'], relative['similarity'], relative['maternal_side'], relative['paternal_side'], None, relative['birth_year'], relative['relationship'], relative['birthplace'], relative['ancestry'])

        # Appending each relative to the user's relative property
        new_user.relatives.append(new_relative)
        models.db_session.add(new_relative)

    # Add the user to the database and commit it
    models.db_session.add(new_user)
    models.db_session.commit()

def getSnps():

    decoded = jwt.decode(request.cookies.get('token'), app.config.get('SECRET_KEY'), algorithms=['HS256'])
    current_user_profile_id = decoded['user_profile_id']

    user_snps = {}

    user_data = models.db_session.query(models.User).filter(models.User.profile_id == current_user_profile_id).first().serialize()
    for user_datum in user_data:
        if user_datum[:2:].lower()=='rs':
            user_snps[user_datum] = user_data[user_datum]

    user_outcomes = []
    for user_snp in user_snps:
        # loop through entire snp table, if any of snp base pairs match up to the base pair in user snps, put in an object with rsid and outcome
        current_snp = models.db_session.query(models.Snp).filter(models.Snp.rs_id == user_snp).filter(models.Snp.dnaPair == user_snps[user_snp]).first()

        if current_snp is not None:
            user_outcomes.append({"title": current_snp.serialize()["title"], "rsid": user_snp, "pair": user_snps[user_snp], "outcome": current_snp.serialize()['outcome'], "video": current_snp.serialize()['video']});

    return jsonify({'outcomes': user_outcomes})

def signup():
    form = SignupForm()
    if form.validate_on_submit():
        # flash('Login requested for OpenID="' + form.openid.data + '", remember_me=' + str(form.remember_me.data))
        user = User()

        user.created_time = datetime.now()
        user.email = form.email.data
        user.nickname = form.username.data
        user.username = form.username.data
        user.password = auth.getPassword(user.username,form.password.data)
        user.role = ROLE_DEFAULT
        user.is_imported = False
        user.status = STATUS_NORMAL
        db.session.add(user)
        db.session.commit()
        auth.login(user,True)
        flash(u'????')
        return redirect('/')

    return render_template('usersignup.html',form=form)

def create_or_login(resp):
    from models import db, User
    username = resp.fullname or resp.nickname or resp.email

    if not username:
        flash('Invalid login. Please try again.', 'danger')
        return redirect(url_for('main.login'))

    user = User.query.filter_by(username=username).first()
    if user is None:
        user = User(username)
        db.session.add(user)
        db.session.commit()

    session['username'] = username
    return redirect(url_for('blog.home'))

def create_or_login(resp):
    from models import db, User
    username = resp.fullname or resp.nickname or resp.email

    if not username:
        flash('Invalid login. Please try again.', 'danger')
        return redirect(url_for('main.login'))

    user = User.query.filter_by(username=username).first()
    if user is None:
        user = User(username)
        db.session.add(user)
        db.session.commit()

    session['username'] = username
    return redirect(url_for('blog.home'))

def create_or_login(resp):
    from models import db, User
    username = resp.fullname or resp.nickname or resp.email

    if not username:
        flash('Invalid login. Please try again.', 'danger')
        return redirect(url_for('main.login'))

    user = User.query.filter_by(username=username).first()
    if user is None:
        user = User(username)
        db.session.add(user)
        db.session.commit()

    session['username'] = username
    return redirect(url_for('blog.home'))

def create_or_login(resp):
    from models import db, User
    username = resp.fullname or resp.nickname or resp.email

    if not username:
        flash('Invalid login. Please try again.', 'danger')
        return redirect(url_for('main.login'))

    user = User.query.filter_by(username=username).first()
    if user is None:
        user = User(username)
        db.session.add(user)
        db.session.commit()

    session['username'] = username
    return redirect(url_for('blog.home'))

def create_or_login(resp):
    from models import db, User
    username = resp.fullname or resp.nickname or resp.email

    if not username:
        flash('Invalid login. Please try again.', 'danger')
        return redirect(url_for('main.login'))

    user = User.query.filter_by(username=username).first()
    if user is None:
        user = User(username)
        db.session.add(user)
        db.session.commit()

    session['username'] = username
    return redirect(url_for('blog.home'))

def create_or_login(resp):
    from models import db, User
    username = resp.fullname or resp.nickname or resp.email

    if not username:
        flash('Invalid login. Please try again.', 'danger')
        return redirect(url_for('main.login'))

    user = User.query.filter_by(username=username).first()
    if user is None:
        user = User(username)
        db.session.add(user)
        db.session.commit()

    session['username'] = username
    return redirect(url_for('blog.home'))

def add_to_history(user_history, sender, receiver, amount, action, finish=False, tx_id="", tip_id=""):
        # convert object to string of name if necessary
        if type(user_history) is models.User:
            user_history = user_history.username

        if tip_id == "":
            tip_id = random.randint(0, 99999999)

        bot_logger.logger.info("Save for history user=%s, sender=%s, receiver=%s, amount=%s, action=%s, finish=%s" % (
            user_history, sender, receiver, amount, action, finish))

        db = TinyDB(config.history_path + user_history + '.json')
        db.insert({
            "id": tip_id,
            "user": user_history,
            "sender": sender,
            "receiver": receiver,
            "amount": amount,
            "action": action,
            "finish": finish,
            "status": "",
            "tx_id": tx_id,
            'time': datetime.datetime.now().isoformat(),
        })
        db.close()

def add_to_history_tip(user_history, action, tip):
        # convert object to string of name if necessary
        if type(user_history) is models.User:
            user_history = user_history.username

        bot_logger.logger.info("Save for history user=%s, sender=%s, receiver=%s, amount=%s, action=%s, finish=%s" % (
            user_history, tip.sender.username, tip.receiver.username, tip.amount, action, tip.finish))

        db = TinyDB(config.history_path + user_history + '.json')
        db.insert({
            "user": user_history,
            "id": tip.id,
            "sender": tip.sender.username,
            "receiver": tip.receiver.username,
            "amount": tip.amount,
            "action": action,
            "finish": tip.finish,
            "status": tip.status,
            "tx_id": tip.tx_id,
            'time': tip.time,
        })
        db.close()

def update_tip(user_history, tip):
        # convert object to string of name if necessary
        if type(user_history) is models.User:
            user_history = user_history.username

        # update only finish tips
        bot_logger.logger.info("update history for user=%s, tip.tx_id=%s" % (user_history, tip.tx_id))
        if tip.id is not None:
            bot_logger.logger.info("update history for user=%s, tip.id=%s" % (user_history, tip.id))

            db = TinyDB(config.history_path + user_history + '.json')
            tip_query = Query()
            db.update({'finish': tip.finish}, tip_query.id == tip.id)
            db.update({'tx_id': tip.tx_id}, tip_query.id == tip.id)
            db.update({'status': tip.status}, tip_query.id == tip.id)
            db.close()
        else:
            bot_logger.logger.warn("update history fail user=%s, tip.id=%s" % (user_history, tip.id))

def update_withdraw(user_history, status, tx_id, tip_id):
        # convert object to string of name if necessary
        if type(user_history) is models.User:
            user_history = user_history.username

        # update only finish tips
        if tip_id is not None:
            bot_logger.logger.info("update history for user=%s, tip.id=%s" % (user_history, tip_id))

            db = TinyDB(config.history_path + user_history + '.json')
            tip_query = Query()
            db.update({'finish': status}, tip_query.id == tip_id)
            db.update({'tx_id': tx_id}, tip_query.id == tip_id)
            db.update({'status': "finish"}, tip_query.id == tip_id)
            db.close()
        else:
            bot_logger.logger.warn("update history fail user=%s, tip.id=%s" % (user_history, tip_id))

def hall_of_fame(msg):
    user = models.User(msg.author.name)
    if user.is_registered():
        message = "Donation Tip to " + config.bot_name + " : "
        donator_list = {}
        hist = models.HistoryStorage.get_user_history(config.bot_name)
        message += "\n\nUser|Donation Ammount\n"
        message += "---|---\n"

        for tip in hist:
            if tip["sender"] in donator_list.keys():
                donator_list[tip["sender"]] = float(donator_list[tip["sender"]]) + tip['amount']
            else:
                donator_list[tip["sender"]] = tip['amount']

        for donor in sorted(donator_list.items(), key=lambda user: user[1], reverse=True):
            message += "%s|%s\n" % (donor[0], str(donor[1]))

        user.send_private_message("Hall Of Fame", message)
    else:
        bot_logger.logger.info('user %s not registered (command : hall_of_fame) ' % user.username)
        msg.reply(Template(lang.message_need_register + lang.message_footer).render(username=user.username))

def donate(msg, tx_queue, failover_time):
    user = models.User(msg.author.name)
    if user.is_registered():
        split_message = msg.body.lower().strip().split()

        donate_index = split_message.index('+donate')
        amount = split_message[donate_index + 1]
        if utils.check_amount_valid(amount) and split_message[donate_index + 2] == 'doge':

            crypto.tip_user(user.username.address, models.User(config.bot_name).address, amount, tx_queue,
                            failover_time)

            models.HistoryStorage.add_to_history(msg.author.name, msg.author.name, config.bot_name, amount, "donate")
        else:
            bot_logger.logger.info(lang.message_invalid_amount)
            user.send_private_message('invalid amount', lang.message_invalid_amount)
    else:
        bot_logger.logger.info('user %s not registered (command : donate) ' % user.username)
        msg.reply(Template(lang.message_need_register + lang.message_footer).render(username=user.username))

def user_command(func):
    def weppar(msg):
        db = models.DBSession()
        user = db.query(models.User).filter_by(wechat_id=msg['FromUserName']).first()
        if user:
            if user.name:
                if 'Content' in msg and msg['Content'][0] == '\\':
                    result = command(msg['Content'][1:], user)
                else:
                    result = func(msg, user, db)
            else:
                user.name = msg['Content'].strip()
                result = f'???????: {user.name}, ?????????? \\name_<?????>'
        else:
            user = models.User(wechat_id=msg['FromUserName'])
            db.add(user)
            result = '???????????'
        db.commit()
        db.close()
        return result

    return weppar

def teams_user(userid):
    user = User.query.get_or_404(userid)
    page = util.as_int(request.values.get('page'), on_fail=1)
    json_data = util.as_int(request.values.get('json'), on_fail=0)

    if json_data:
        teams_dict = {}
        for team in user.teams:
            team_dict = {}
            team_dict['name'] = team.name
            team_dict['tag'] = team.tag
            team_dict['flag'] = team.flag
            team_dict['logo'] = team.logo
            team_dict['players'] = filter(lambda x: bool(x), team.auths)
            teams_dict[team.id] = team_dict
        return jsonify(teams_dict)

    else:
        # Render teams page
        my_teams = (g.user is not None and userid == g.user.id)
        teams = user.teams.paginate(page, 20)
        return render_template(
            'teams.html', user=g.user, teams=teams, my_teams=my_teams,
                               page=page, owner=user)

def initdb(username, password):
    db.create_all()
    import binascii
    u = models.User(email=username, password_hash=bcrypt.generate_password_hash(binascii.hexlify(password)), role=0, status=1)
    db.session.add(u)
    db.session.commit()
    print 'Database initialized.'
    # remove below for production
    t = models.Target(name='demo', guid='aedc4c63-8d13-4a22-81c5-d52d32293867')
    db.session.add(t)
    db.session.commit()
    b = models.Beacon(target_guid='aedc4c63-8d13-4a22-81c5-d52d32293867', agent='HTML', ip='1.2.3.4', port='80', useragent='Mac OS X', comment='this is a comment.', lat='38.2531419', lng='-85.7564855', acc='5')
    db.session.add(b)
    db.session.commit()
    b = models.Beacon(target_guid='aedc4c63-8d13-4a22-81c5-d52d32293867', agent='HTML', ip='5.6.7.8', port='80', useragent='Mac OS X', comment='this is a comment.', lat='34.855117', lng='-82.114192', acc='1')
    db.session.add(b)
    db.session.commit()

def render_home(self, **extra):
        from models import User

        user_id = self.get_secure_cookie('user_id')

        if user_id:
            user = session.query(User).get(int(user_id))
        else:
            user = None

        context = common_context(
            settings.SOCIAL_AUTH_AUTHENTICATION_BACKENDS,
            load_strategy(self),
            user=user,
            plus_id=getattr(settings, 'SOCIAL_AUTH_GOOGLE_PLUS_KEY', None),
            **extra
        )
        self.render('home.html', **context)

def signup():
    email, pw = strip_creds_from_req(request.data)

    if not email or not pw:
        return rh.error(**ec.INCOMPLETE_LOGIN_CREDS)

    user = find(User, dict(email=email))
    if user: return rh.error(**ec.EMAIL_TAKEN)

    try:
        user = create(User, dict(email=email, password=pw))
        session = create(Session, dict(user_id=user.id))
        return rh.json_response(with_cookie=(cookie_name, session.token))
    except Exception as e:
        app.logger.error('Error signing up new user with email: {}, with error: {}'.format(email, e))
        return rh.error(**ec.USER_SIGNUP_ERROR)

def signup():
  if 'email' in session:
    return redirect(url_for('home'))

  form = SignupForm()

  if request.method == "POST":
    if form.validate() == False:
      return render_template('signup.html', form=form)
    else:
      newuser = User(form.first_name.data, form.last_name.data, form.email.data, form.password.data)
      db.session.add(newuser)
      db.session.commit()

      session['email'] = newuser.email
      return redirect(url_for('home'))

  elif request.method == "GET":
    return render_template('signup.html', form=form)

def get_user_profile_info(token):
    """
    Calls the get current user's profile endpoint on Spotify.
    :param token: access token attained from Spotify on behalf of user
    :return: a User object
    """
    user_profile_endpoint = 'https://api.spotify.com/v1/me'
    headers = {'Authorization': 'Bearer %s' % token}

    r = requests.get(user_profile_endpoint, headers=headers)
    if r.status_code != 200:
        return None
    profile = r.json()

    display_name = profile['display_name']
    profile_id = profile['id']
    email = profile['email']

    return User(display_name=display_name,
                profile_id=profile_id,
                email=email)

def authorized(oauth_token):
    """
    authorized: update user table and login session
    Args:
        oauth_token (str): access token from oauth server
    Returns:
        return homepage
    """
    state = request.args.get('state')
    if state != login_session['state']:
        response = make_response(json.dumps('Invalid state parameter.'), 401)
        response.headers['Content-Type'] = 'application/json'
        return response

    login_session['access_token'] = oauth_token

    data = github.get('user')
    email = data['email']
    name = data['name']

    user = session.query(User).filter_by(email=email).first()
    if not user:
        user = User(name=name, email=email)
        session.add(user)
    user.name = name
    session.commit()
    login_session['user_id'] = user.id

    flash("Logged in as %s!" % name)
    return redirect(url_for('show_home'))

def root():
    users = db.session.query(User).all()
    return u"<br>".join([u"{0}: {1}".format(user.name, user.email) for user in users])