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

def post(self):
        '''Endpoint for getting auth token for existing user'''
        parser = reqparse.RequestParser()
        parser.add_argument('email', required=True)
        parser.add_argument('password', required=True)
        args = parser.parse_args()
        hash_obj = hashlib.sha256()
        hash_obj.update(args.password.encode('utf-8'))
        password_hash = hash_obj.hexdigest()
        user = db.session.query(User).filter((User.email==args.email) & (User.password_hash==password_hash)).first()
        if user is None:
            return redirect('/?message=%s' % 'Could not find account.')

        # allocate and maintain session token
        token = os.urandom(256)
        tokens[token] = args.email
        session['token'] = token
        session['username'] = user.username
        return redirect('/')

def request_loader(request):
    email = request.form.get('email')

    if email is None:
        return

    password = request.form['pw']

    loaded_user = None

    user = db.session.query(User).filter(User.email == email).first()

    if user is None:
        return

    user.is_authenticated = user.check_password(password)
    loaded_user = user

    return loaded_user

def login():

    if flask.request.method == 'GET':
        return render_template('login.html')

    email = request.form['email']
    password = request.form['pass']

    for user in db.session.query(User).filter(User.email == email):
        print('Login Result: ' + user.email, file=sys.stderr)

        if user.check_password(password):
            flask_login.login_user(user)
            return redirect('dashboard')

    return redirect('login_failed')

def register():
    if request.method == 'POST':
        email = request.form['email']
        password = request.form['password']
        phone = flask.request.form['phone']

        if email is not None and password is not None and phone is not None:
            user = User(email, password, phone)
            db.session.add(user)
            db.session.commit()

            flags = Flags(user.id, False, False, False)
            pi = Pi(user.id, datetime.utcnow())

            db.session.add(flags)
            db.session.add(pi)
            db.session.commit()

            flask_login.login_user(user)

            return redirect('dashboard')
        else:
            print('Error creating user: ' + str(email), file=sys.stderr)

    return render_template('register.html')

def post(self):
        '''Endpoint for registering and getting auth token for new user'''
        parser = reqparse.RequestParser()
        parser.add_argument('username', required=True)
        parser.add_argument('email', required=True)
        parser.add_argument('password', required=True)
        args = parser.parse_args()

        # check username and email uniqueness
        duplicate = db.session.query(User).filter((User.username==args.username) | (User.email==args.email)).first()
        if duplicate is not None:
            return redirect('/?message=%s' % 'Account already exists for this email.')

        hash_obj = hashlib.sha256()
        hash_obj.update(args.password.encode('utf-8'))
        user_obj = User(
            username=args.username,
            email=args.email,
            password_hash=hash_obj.hexdigest()
        )
        db.session.add(user_obj)
        db.session.commit()

        # allocate and maintain session token
        token = os.urandom(256)
        tokens[token] = args.email
        session['token'] = token
        session['username'] = args.username
        return redirect('/')

def email_exists(form,field):
    if User.select().where(User.email == field.data).exists():
        raise ValidationError('User with this email already exists.')

def is_valid(self):
        if len(self.form['username']) < 4:
            return self.error('username must be at least 4 characters')
        if self.form['username'] == 'anonymous':
            return self.error('username cannot be anonymous')
        if len(self.form['password']) < 8:
            return self.error('password must be at least 8 characters')
        if User.select().where(User.username == self.form['username']).count():
            return self.error('username is already in use')
        if self.form['email']:
            if not validate_email(self.form['email']):
                return self.error('email is not valid')
            if User.select().where(User.email == self.form['email']).count():
                return self.error('email is already in use')
        return super(RegistrationValidator, self).is_valid()

def is_valid(self):
        if not validate_email(self.form['email']):
            return self.error('email is not valid')
        return super(ResetValidator, self).is_valid()

def user_loader(email):
    users = db.session.query(User).filter(User.email == email)

    if users is None:
        return

    temp_user = None

    for user in users:
        temp_user = user

    if temp_user is None:
        return

    return temp_user

def settings():
    if request.method == 'POST':
        email = request.form['email']
        curpassword = request.form['curpassword']
        password = request.form['password']
        phone = flask.request.form['phone']
        pi_id = flask.request.form['piid']

        room_name = flask.request.form['room_name']
        capture_framerate = flask.request.form['capture_framerate']
        output_framerate = flask.request.form['output_framerate']
        threshold_frame_count = flask.request.form['threshold_frame_count']

        current_user = User.query.filter_by(id=flask_login.current_user.id).first()

        if pi_id == 'true':
           current_user.pi_id = str(uuid.uuid4())

        if current_user.check_password(curpassword) and password is not '':
            current_user.passhash = generate_password_hash(password)

        current_user.email = email
        current_user.phone = phone

        pi_data = Pi.query.filter_by(id=flask_login.current_user.id).first()
        pi_data.room_name = room_name
        pi_data.capture_framerate = capture_framerate
        pi_data.output_framerate =output_framerate
        pi_data.threshold_frame_count = threshold_frame_count

        flag=Flags.query.filter_by(id=flask_login.current_user.id).first()
        flag.request_update_settings = True

        db.session.commit()

        return redirect('settings')

    return render_template('settings.html', user_data = flask_login.current_user, pi_data = Pi.query.filter_by(id=flask_login.current_user.id).first())

def email_exists(form, field):
    if User.select().where(User.email == field.data).exists():
        raise ValidationError('User with that email already exists.')

def sign_up():
    form = SignUpForm()
    user = User()
    if form.validate_on_submit():
        user_name = request.form.get('user_name')
        user_email = request.form.get('user_email')
        password = request.form.get('password')
        password = generate_password_hash(password)
        register_check = User.query.filter(db.or_(
            User.nickname == user_name, User.email == user_email)).first()
        if register_check:
            flash("????????")
            return redirect('/sign-up')

        if len(user_name) and len(user_email):
            user.nickname = user_name
            user.email = user_email
            user.role = ROLE_USER
            user.password = password
            try:
                db.session.add(user)
                db.session.commit()
            except:
                flash("???????????")
                return redirect('/sign-up')

            flash("????")
            return redirect('/login')

    return render_template("sign_up.html", form=form)

def email_exists(form, field):
    if User.select().where(User.email == field.data).exists():
        raise ValidationError('User with that email already exists.')

def post(self, task_id):
        ''' Endpoint for submitting code '''
        parser = reqparse.RequestParser()
        parser.add_argument('language', required=True)
        parser.add_argument('code', required=True)
        args = parser.parse_args()
        task = manager.get_task(task_id)
        if task_id is None or task is None:  # no task exists with the stated id
            return redirect('/?message=%s' % 'task does not exist.')
        test_cases = task.get('test_cases', None)
        # get user id
        if 'token' not in session:
            try:
                session.pop('username')
            except KeyError: pass
            return redirect('/?message=%s' % 'no auth token supplied.')
        if session['token'] not in tokens:
            try:
                session.pop('username')
            except KeyError: pass
            session.pop('token')
            return redirect('/?message=%s' % 'invalid token supplied.')
        user = db.session.query(User).filter(User.email==tokens[session['token']]).first()

        # verify response using same input
        result = docker_verify(args.code, args.language, test_cases)
        if result is None:
            return redirect('/?task=%d&message=%s' % (task_id, 'language is not supported'))
        if all(result):
            if(len(args.code) < best_answer[task_id][args.language]):
                best_answer[task_id][args.language] = len(args.code)
            max_points = int(task.get('points', None))
            points = max_points * best_answer[task_id][args.language]/len(args.code)

            # look for better answer from same user
            prev_answer = db.session.query(Answer).filter((Answer.user_id==user.id) & (Answer.task_id==task_id) & (Answer.points>points)).first()
            if prev_answer is None:
                answer = Answer(
                    user_id=user.id,
                    task_id=task_id,
                    length=len(args.code),
                    points=points,
                    language=args.language
                )
                db.session.add(answer)
                user.points = sum([answer.points for answer in user.answers])
                db.session.commit()
            return redirect('/?task=%d&message=%s' % (task_id, 'answer judged correct.'))
        return redirect('/?task=%d&message=%s' % (task_id, 'answer judged incorrect.'))