Python flask_login.current_user 模块，is_anonymous() 实例源码

def event_list():
    form = AddUserToEventForm(request.form)
    if request.method == 'POST' and form.validate():
        if current_user.is_anonymous or not current_user.is_authenticated:
            flash('You must log in to register for events.')
            return redirect(url_for('events.event_list'))

        event = Event.query.filter(Event.id == form.event_id.data).first()
        if current_user in event.users:
            flash('You are already registered for this event!')
            return redirect(url_for('events.event_list'))
        else:
            event.users.append(current_user)
            db_session.add(event)
            db_session.commit()
            flash('Registration successful!')
            return redirect(url_for('events.event_list'))
    else:
        events = Event.query.limit(10)
        return render_template('events.html', events=events, form=form)

def check_auth(func):
    """
    This decorator for routes checks that the user is authorized (or that no login is required).
    If they haven't, their intended destination is stored and they're sent to get authorized.
    It has to be placed AFTER @app.route() so that it can capture `request.path`.
    """
    if 'login' not in conf:
        return func
    # inspired by <https://flask-login.readthedocs.org/en/latest/_modules/flask_login.html#login_required>
    @functools.wraps(func)
    def decorated_view(*args, **kwargs):
        if current_user.is_anonymous:
            print('unauthorized user visited {!r}'.format(request.path))
            session['original_destination'] = request.path
            return redirect(url_for('get_authorized'))
        print('{} visited {!r}'.format(current_user.email, request.path))
        assert current_user.email.lower() in conf.login['whitelist'], current_user
        return func(*args, **kwargs)
    return decorated_view

def password_reset_request():
    """Request for reseting password when user forget his/her password.
    """
    if not current_user.is_anonymous:
        return redirect(url_for('.index'))
    form = PasswordResetRequestForm()
    if form.validate_on_submit():
        user = User.query.filter_by(email=form.email.data).first()
        if user:
            token = user.generate_reset_token()
            send_email(user.email, 'Reset Your Password',
                       'auth/email/reset_password',
                       user=user, token=token,
                       next=request.args.get('next'))
        flash('An email with instructions to reset your password has been '
              'sent to you.')
        return redirect(url_for('auth.login'))
    return render_template('auth/reset_password.html', form=form)

def password_reset(token):
    """Reset password using token.
    """
    if not current_user.is_anonymous:
        return redirect(url_for('.index'))
    form = PasswordResetForm()
    if form.validate_on_submit():
        user = User.query.filter_by(email=form.email.data).first()
        if user is None:
            return redirect(url_for('.index'))
        if user.reset_password(token, form.password.data):
            flash('Your password has been updated.')
            return redirect(url_for('auth.login'))
        else:
            return redirect(url_for('.index'))
    return render_template('auth/reset_password.html', form=form)

def show_preloader_start_authentication():

    if not current_user.is_anonymous:
        return redirect(url_for('index'))


    # store in the session id of the asynchronous operation
    status_pending = AsyncOperationStatus.query.filter_by(code='pending').first()
    async_operation = AsyncOperation(async_operation_status_id=status_pending.id)
    db.session.add(async_operation)
    db.session.commit()
    # store in a session the id of asynchronous operation
    session['async_operation_id'] = str(async_operation.id)

    taskman.add_task(external_auth)

    return redirect(url_for('preloader'))

def unconfirmed() :
    if current_user.is_anonymous() or current_user.confirmed :
        return redirect(url_for('auth.unconfirmed')
    return render_template('auth/unconfirmed.html')


#??????????
@auth.route('/confirm')
@login_required
def resend_confirmation():
    token = current_user.generate_confirmation_token()
    send_email(current_user.email,'Confim Your Accout' , 'auth/email/confirm',user=current_user,token=token)
    flash('A new confirmation email has been sent to you by email.')
    return redirect(url_for('main.index'))


# ???????
@main.route('/user/<username>')
def user(username) :
    user = User.query.filter_by(username=username).first()
    if user is None :
        abort(404) 
    return render_template('user.html',user=user)

def get_navbar():
    navbar = Navbar('', *[
        View('Href Today', 'dashboard.get_links'),
    ])
    navbar.logo = Logo('images/logo.png', 'dashboard.get_links')
    if current_user.is_anonymous:
        navbar.right_side_items = [View('Login', 'user.login')]
    else:
        navbar.right_side_items = [
            Subgroup(
                'Account',
                View('Change Password', 'user.change_password'),
                View('Logout', 'user.logout'),
            )
        ]
    return navbar

def add_training():
    if current_user.is_anonymous:
        return "Sorry but you must be logged in to add training data."
    mac = request.form['mac']
    location = request.form['location']
    current_detection = Detection.query.filter_by(type='detection', mac=mac).first()
    training_detection = TrainingDetection(location=Location.query.filter_by(value=location).first(), mac=mac)
    db.session.add(training_detection)
    for m in current_detection.measurements:
        power = m.power
        if measurement_too_old(m):
            power = -100
        measurement = Measurement(m.slave_id, power, datetime.datetime.now(), training_detection)
        db.session.add(measurement)
    db.session.commit()
    user = User.query.filter_by(email=current_user.email).first()
    if not Device.query.filter_by(user=user, mac=mac).first():
        db.session.add(Device(mac=mac, user=user))
        db.session.commit()
    return redirect(url_for('training'))

def password_reset_request():
    if not current_user.is_anonymous:
        return redirect(url_for('home.index'))
    form = PasswordResetRequestForm()
    if form.validate_on_submit():
        me = models.Users.query.filter_by(email=form.email.data).first()
        if me:
            token = me.generate_reset_token()
            send_email(me.email, 'Reset Your Password',
                       'mail/user/reset_password',
                       user=me, token=token,
                       next=request.args.get('next'))
        flash('An email with instructions to reset your password has been '
              'sent to you.')
        return redirect(url_for('user.login'))
    return render_template('user/reset_password.html', form=form)

def password_reset_request():
    if not current_user.is_anonymous:
        return redirect(url_for('main.index'))
    form = PasswordResetRequestForm()
    if form.validate_on_submit():
        user = User.query.filter_by(email=form.email.data).first()
        if user:
            token = user.generate_reset_token()
            send_email(user.email, 'Reset Your Password',
                       'auth/email/reset_password',
                       user=user, token=token,
                       next=request.args.get('next'))
        flash('An email with instructions to reset your password has been'
              'sent to you.')
        return redirect(url_for('auth.login'))
    return render_template('auth/reset_password.html', form=form)

def index():
    events = None
    user = None
    if not current_user.is_anonymous:
        user = current_user
        events = user.events

    return render_template('index.html', events=events, user=user)

def event_detail(event_id):
    event = Event.query.filter(Event.id == event_id).first()
    form = PhotoForm(CombinedMultiDict((request.files, request.form)))
    user_photo = next((p for p in event.photos if p.user_id ==
                       current_user.id), None) if not current_user.is_anonymous else None
    if request.method == 'POST':
        if user_photo != None:
            flash('You have already uploaded a photo to this event.')
            return redirect(url_for('events.event_detail', event_id=event.id))
        return event_upload(request, event, form)

    winner = Photo.query.filter(
        Photo.id == event.winner_id).first() if event.winner_id else None
    print(winner)
    return render_template('event_detail.html', event=event, form=form, photo=user_photo, winner=winner)

def event_upload(request, event, form):
    if not form.validate():
        print('upload failed')
        flash('Upload failed.')
        return redirect(url_for('events.event_detail', event_id=event.id))
    if current_user.is_anonymous:
        print('user not logged in')
        flash('You must be logged in to do this.')
        return redirect(url_for('events.event_detail', event_id=event.id))
    if current_user not in event.users:
        flash('You must be registered for this event to do this.')
        return redirect(url_for('events.event_detail', event_id=event.id))

    photo = form.photo.data
    filename = secure_filename(form.photo.data.filename)
    filename = current_user.username + '_' + filename
    try:
        abs_filename = getcwd() + '/luminance/static/photos/' + filename
        form.photo.data.save(abs_filename)
        p = Photo(url='/static/photos/' + filename)
        current_user.photos.append(p)
        current_user.exp += 1
        event.photos.append(p)
        db_session.add(p)
        db_session.add(current_user)
        db_session.add(event)
        db_session.commit()
    except Exception:
        print_exc()
        flash('Upload failed.')
        return redirect(url_for('events.event_list'))

    flash('Upload successful!')
    return redirect(url_for('events.event_list'))

def get_authorized():
        "This route tries to be clever and handle lots of situations."
        if current_user.is_anonymous:
            return google_sign_in.authorize()
        else:
            if 'original_destination' in session:
                orig_dest = session['original_destination']
                del session['original_destination'] # We don't want old destinations hanging around.  If this leads to problems with re-opening windows, disable this line.
            else:
                orig_dest = url_for('homepage')
            return redirect(orig_dest)

def ask_login():
  if not current_user.is_anonymous:
    return

  if getattr(current_app.view_functions[request.endpoint], 'no_login', False):
    return

  return redirect(url_for('ares.aresLogin', next=request.endpoint))

def ask_login():
  if not current_user.is_anonymous:
    return

  if getattr(current_app.view_functions[request.endpoint], 'no_login', False):
    return

  return redirect(url_for('ares.aresLogin', next=request.endpoint))

def is_accessible(self):
        return (
            not AUTHENTICATE or (
                not current_user.is_anonymous() and
                current_user.is_authenticated()
            )
        )

def is_accessible(self):
        return (
            not AUTHENTICATE or
            (not current_user.is_anonymous() and current_user.is_superuser())
        )

def is_accessible(self):
        return (
            not AUTHENTICATE or
            (not current_user.is_anonymous() and current_user.data_profiling())
        )

def validate(self):
        check_validate = super(ContactForm, self).validate()

        if not check_validate:
            return False

        if current_user.is_anonymous() and self.checksum.data != '4':
            self.checksum.errors.append('The answer is: 4')
            return False

        return True

def main(page=1, page_size=24):
    ebooks=None
    if not current_user.is_anonymous and current_user.has_role('user'):
        ebooks=model.Ebook.query.order_by(desc(model.Ebook.created)).paginate(page, page_size)

    return render_template('main.html', ebooks=ebooks)

def save_questionnaire_store(func):
    def save_questionnaire_store_wrapper(*args, **kwargs):
        response = func(*args, **kwargs)
        if not current_user.is_anonymous:
            questionnaire_store = get_questionnaire_store(current_user.user_id, current_user.user_ik)

            questionnaire_store.add_or_update()

        return response

    return save_questionnaire_store_wrapper

def is_it_active():
    '''
    If Ip is Active
    :param:

    :return:   [JSON]
        "status": 0,
        "message": "User Not Login!",
        "data": {}
    '''
    if current_user.is_anonymous():
        return jsonify({
            "status": 0,
            "message": "User Not Login!",
            "data": {}
        })
    else :
        return jsonify({
            "status": 1,
            "message": "User has Login!",
            "data": {
                "id"       : current_user.userID,
                "nickname" : current_user.nickName,
                "isAuth"   : current_user.isAuthenticated,
                "username" : current_user.userName,
                "email"    : current_user.email,
                "qq"       : current_user.qq,
                "compressPicture": current_user.compressPicture
            }
        })

def facebook_authorize():
    if not current_user.is_anonymous:
        return redirect(url_for('index'))
    oauth = FacebookSignIn()
    return oauth.authorize()


# returns status of the async operation

def before_dashboard_request():
    if current_user.is_anonymous:
        return redirect(url_for('user.login'))
    g.draft_links_count = get_draft_links_count()

def is_belongs_to_me(obj):
    return obj and not current_user.is_anonymous and obj.user_id == current_user.id

def before_request():
    if current_user.is_authenticated() and current_user.is_anonymous() is False:
        current_user.ping()
        if not current_user.confirmed and request.endpoint and request.endpoint[:4] == 'view' and request.endpoint != 'static':
            return redirect(url_for('unconfirmed'))

def password_reset(token):
    if not current_user.is_anonymous:
        return redirect(url_for('view_rents'))
    form = PasswordResetForm()
    if form.validate_on_submit():
        user = orm.User.query.filter_by(email=form.email.data).first()
        if user is None:
            return redirect(url_for('view_rents'))
        if user.reset_password(token, form.password.data):
            flash('????????.')
            return redirect(url_for('login'))
        else:
            return redirect(url_for('view_rents'))
    return render_template('auth/reset_password.html', form=form)

def password_reset_request():
    if not current_user.is_anonymous:
        return redirect(url_for('view_rent'))
    form = PasswordResetRequestForm()
    if form.validate_on_submit():
        user = orm.User.query.filter_by(email=form.email.data).first()
        if user:
            token = user.generate_reset_token()
            send_email(user.email, u'????',
                       'auth/email/reset_password',
                       user=user, token=token,
                       next=request.args.get('next'))
        flash(u'???????????????????')
        return redirect(url_for('login'))
    return render_template('auth/reset_password.html', form=form)

def unconfirmed():
    if current_user.is_anonymous() is True or current_user.confirmed:
        return redirect(url_for('view_rents'))
    form = PageInfo()
    logic.LoadBasePageInfo('?????', form)
    return render_template('auth/unconfirmed.html', form=form)

def add_device():
    if current_user.is_anonymous:
        return "Sorry but you must be logged in to add devices."
    mac = request.form['mac'].upper()
    email = request.form['email'].lower()
    user = User.query.filter_by(email=email).first()
    if not Device.query.filter_by(user=user, mac=mac).first():
        db.session.add(Device(mac=mac, user=user))
        db.session.commit()
    return redirect(url_for('index'))

def is_accessible(self):
        return (
            not AUTHENTICATE or (
                not current_user.is_anonymous() and
                current_user.is_authenticated()
            )
        )

def is_accessible(self):
        return (
            not AUTHENTICATE or
            (not current_user.is_anonymous() and current_user.is_superuser())
        )

def is_accessible(self):
        return (
            not AUTHENTICATE or
            (not current_user.is_anonymous() and current_user.data_profiling())
        )

def unconfirmed():
    if current_user.is_anonymous or current_user.confirmed:
        return redirect(url_for('home.index'))
    return render_template('user/unconfirmed.html')

def password_reset(token):
    if not current_user.is_anonymous:
        return redirect(url_for('home.index'))
    form = PasswordResetForm()
    if form.validate_on_submit():
        me = models.Users.query.filter_by(email=form.email.data).first()
        if me is None:
            return redirect(url_for('home.index'))
        if me.reset_password(token, form.password0.data):
            flash('Your password has been updated.')
            return redirect(url_for('user.login'))
        else:
            return redirect(url_for('home.index'))
    return render_template('user/reset_password.html', form=form)

def unconfirmed():
    if current_user.is_anonymous or current_user.confirmed:
        return redirect(url_for('main.index'))
    return render_template('auth/unconfirmed.html', user=current_user)

def unconfirmed():
    if current_user.is_anonymous or current_user.confirmed:
        return redirect(url_for('main.index'))
    return render_template('auth/unconfirmed.html')

def password_reset(token):
    if not current_user.is_anonymous:
        return redirect(url_for('main.index'))
    form = PasswordResetForm()
    if form.validate_on_submit():
        user = User.query.filter_by(email=form.email.data).fifirst()
        if user is None:
            return redirect(url_for('main.index'))
        if user.reset_password(token, form.password.data):
            flash('Your password has been updated.')
            return redirect(url_for('auth.login'))
        else:
            return redirect(url_for('main.index'))
    return render_template('auth/reset_password.html', form=form)

def unconfirmed():
    if current_user.is_anonymous or current_user.confirmed:
        return redirect(url_for('main.index'))
    return render_template('auth/unconfirmed.html')

def password_reset_request():
    if not current_user.is_anonymous:
        return redirect(url_for('main.index'))
    form = PasswordRequestForm()
    if form.validate_on_submit():
        user = User.query.filter_by(email = form.email.data).first()
        if user:
            token = user.generate_reset_token()
            send_email(user.email, 'Reset Your Password', 'auth/email/reset_password', user = user, token = token, next = request.args.get('next'))
        flash('An email with instrcuctions to reset your password has been sent to you')
        return redirect(url_for('auth.login'))
    return render_template('auth/reset_password.html', form = form)

def password_reset(token):
    if not current_user.is_anonymous:
        return redirect(url_for('main.index'))
    form = PasswordResetForm()
    if form.validate_on_submit():
        user = User.query.filter_by(email = form.email.data).first()
        if user is None:
            flash('Invalid email address')
            return redirect(url_for('main.index'))
        if user.reset_password(token, form.password.data):
            flash('Your password has been updated')
            return redirect(url_for('auth.login'))
        else:
            return redirect(url_for('main.index'))
    return render_template('auth/reset_password.html', form = form)

def unconfirmed():
    if current_user.is_anonymous or current_user.confirmed:     #???????????????
        return redirect(url_for('main.index'))
    return render_template('auth/unconfirmed.html')

def unconfirmed():
    if current_user.is_anonymous or current_user.confirmed:
        return redirect(url_for ('main.index'))
    return render_template('auth/unconfirmed.html')

def password_reset_request():
    if not current_user.is_anonymous:
        return redirect(url_for('main.index'))
    form = PasswordResetRequestForm()
    if form.validate_on_submit():
        user = User.query.filter_by(email=form.email.data).first()
        if user:
            token = user.generate_reset_token()
            send_email(user.email, 'Reset Your Password',
                    'auth/email/reset_password',
                    user=user, token=token,
                    next=request.args.get('next'))
        flash(u'?????????????????')
        return redirect(url_for('auth.login'))
    return render_template('auth/reset_password.html', form=form)

def password_reset(token):
    if not current_user.is_anonymous:
        return redirect(url_for('main.index'))
    form = PasswordResetForm()
    if form.validate_on_submit():
        user = User.query.filter_by(email=form.email.data).first()
        if user is None:
            return redirect(url_for('main.index'))
        if user.reset_password(token, form.password.data):
            flash(u'??????')
            return redirect(url_for('auth.login'))
        else:
            return redirect(url_for('main.index'))
    return render_template('auth/reset_password.html', form=form)

def unconfirmed():
    if current_user.is_anonymous or current_user.confirmed:
        return redirect(url_for('main.index'))
    return render_template('auth/unconfirmed.html')



#????

def password_reset_request():
    if not current_user.is_anonymous:
        return redirect(url_for('main.index'))
    form = ResetPasswordForm()
    if form.validate_on_submit():
        user = User.query.filter_by(email = form.email.data).first()
        if user:
            token = user.generate_reset_token()
            send_email(user.email,'Reset Your Password','auth/email/reset_password',user=user,token=token,next=request.args.get('next'))

        flash('An email with instructions to reset your password has been sent to you')
        return redirect(url_for('auth.login'))
    return render_template('auth/reset_password.html',form=form)

def password_reset(token):
    if not current_user.is_anonymous:
        return redirect(url_for('main.index'))
    form = ResetPasswordForm()
    if form.validate_on_submit():
        user = User.query.filter_by(email=form.email.data).first()
        if user is None:
            return redirect(url_for('main.index'))
        if user.reset_password(token,form.password.data):
            flash('Your password has ben updated')
            return redirect(url_for('auth.login'))
        else:
            return redirect(url_for('main.index'))
    return render_template('auth/reset_password.html',form=form)