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

def adjust_status_submit(club_type):
    '''Input change in activeness into database'''
    ids = request.form.getlist('activeness')
    clubs = map(Club, ids)
    for club in clubs:
        if club.is_active:
            club.is_active = False
        else:
            club.is_active = True
    ids = request.form.getlist('joinmode')
    clubs = map(Club, ids)
    for club in clubs:
        if club.joinmode == ClubJoinMode.FREE_JOIN:
            club.joinmode = ClubJoinMode.BY_INVITATION
        elif club.joinmode == ClubJoinMode.BY_INVITATION:
            club.joinmode = ClubJoinMode.FREE_JOIN
    flash('The change in clubs\' status has been submitted.', 'adjust_status')
    return redirect(url_for('.adjust_status', club_type=club_type))

def allclubsinfo():
    '''Allow admin to download all clubs' info'''
    info = []
    info.append(('Club ID', 'Name', 'Leader', 'Leader\'s Class', 'Teacher',
                 'Introduction', 'Location', '# of Members',
                 '# of registered activities', '# of activities with attendance',
                 'Is Active or Not', 'Type', 'Description'))
    info.extend([(club.id, club.name, club.leader.passportname,
                  club.leader.grade_and_class,
                  club.teacher.email, club.intro, club.location,
                  len(club.members), len(club.activities()),
                  len(club.activities(require_attend=True)),
                  str(club.is_active),
                  club.type.format_name, club.description.raw)
                 for club in Club.allclubs()])

    return download_xlsx('All Clubs\' Info.xlsx', info)

def test_flask_login_user_login(app):
    # LoginManager could be set up in app fixture in conftest.py instead
    login_manager = LoginManager()
    login_manager.init_app(app)

    # TODO: once event is marked, user id exists in MonthEvents and test will
    #       continue to pass, regardless of continued success; set to current
    #       microsecond to temporarily circumvent, but there should be a better
    #       way to fix user_id assignment (or tear down redis or something)
    user_id = datetime.now().microsecond

    with app.test_request_context():
        # set up and log in user
        user = User()
        user.id = user_id
        login_user(user)

        # test that user was logged in
        assert current_user.is_active
        assert current_user.is_authenticated
        assert current_user == user

        # test that user id was marked with 'user:logged_in' event
        assert user_id in MonthEvents('user:logged_in', now.year, now.month)

def action_reset(self, ids):
        """Reset batch task records."""
        status = "The record was reset at " + datetime.now().isoformat(timespec="seconds")
        with db.atomic():
            try:
                count = self.model.update(processed_at=None, status=status).where(
                    self.model.is_active, self.model.processed_at.is_null(False),
                    self.model.id.in_(ids)).execute()

                if self.model == FundingRecord:
                    count = FundingContributor.update(processed_at=None, status=status).where(
                        FundingContributor.funding_record.in_(ids),
                        FundingContributor.status.is_null(False)).execute()

            except Exception as ex:
                db.rollback()
                flash(f"Failed to activate the selected records: {ex}")
                app.logger.exception("Failed to activate the selected records")

            else:
                if self.model == FundingRecord:
                    flash(f"{count} Funding Contributor records were reset for batch processing.")
                else:
                    flash(f"{count} Affiliation records were reset for batch processing.")

def action_reset(self, ids):
        """Batch reset of users."""
        try:
            status = " The record was reset at " + datetime.now().isoformat(timespec="seconds")
            count = self.model.update(
                processed_at=None, status=status).where(
                    self.model.status.is_null(False), self.model.id.in_(ids)).execute()
            funding_record_id = self.model.select().where(
                self.model.id.in_(ids))[0].funding_record_id
            FundingRecord.update(
                processed_at=None, status=FundingRecord.status + status).where(
                    FundingRecord.is_active, FundingRecord.id == funding_record_id).execute()
        except Exception as ex:
            flash(f"Failed to activate the selected records: {ex}")
            app.logger.exception("Failed to activate the selected records")

        else:
            flash(f"{count} Funding Contributor records were reset for batch processing.")

def activate_all():
    """Batch registraion of users."""
    _url = request.args.get("url") or request.referrer
    task_id = request.form.get('task_id')
    task = Task.get(id=task_id)
    try:
        if task.task_type == 0:
            count = AffiliationRecord.update(is_active=True).where(
                AffiliationRecord.task_id == task_id,
                AffiliationRecord.is_active == False).execute()  # noqa: E712
        elif task.task_type == 1:
            count = FundingRecord.update(is_active=True).where(
                FundingRecord.task_id == task_id,
                FundingRecord.is_active == False).execute()  # noqa: E712
    except Exception as ex:
        flash(f"Failed to activate the selected records: {ex}")
        app.logger.exception("Failed to activate the selected records")
    else:
        flash(f"{count} records were activated for batch processing.")
    return redirect(_url)

def clubintro(club):
    '''Club Intro'''
    free_join = (current_user.is_active and
                 club.joinmode == ClubJoinMode.FREE_JOIN and
                 current_user.type == UserType.STUDENT and
                 current_user not in club.members)
    see_email = (current_user.is_active and
                 current_user.type == UserType.ADMIN or
                 current_user == club.leader)
    return render_template('club/clubintro.jinja2',
                           free_join=free_join,
                           see_email=see_email)

def test_flask_login_user_logout(app):
    login_manager = LoginManager()
    login_manager.init_app(app)

    user_id = datetime.now().microsecond

    with app.test_request_context():
        # set up, log in, and log out user
        user = User()
        user.id = user_id
        login_user(user)
        logout_user()

        # test that user was logged out
        assert not current_user.is_active
        assert not current_user.is_authenticated
        assert not current_user == user

        # test that user id was marked with 'user:logged_out' event
        assert user_id in MonthEvents('user:logged_out', now.year, now.month)


# SQLALCHEMY

# TODO: Instead of sqlalchemy fixture (return: db, User),
#       each test could use sqlalchemy fixture (return:
#       db) and sqlalchemy_user fixture (return: User);
#       tests should use whichever is better practice.

def test_noauth_user_mixin_is_authenticated(self, testapp):

        assert current_user is not None
        assert current_user.is_authenticated
        assert current_user.is_active
        assert not current_user.username  # no username for this faux user

def get_session_expired():
    if current_user.is_active:
        remove_survey_session_data()

    return render_template('session-expired.html')

def get_sign_out():
    if current_user.is_active:
        remove_survey_session_data()

    return render_template('signed-out.html')

def is_accessible(self):
        """Verify if the view is accessible for the current user."""
        if not current_user.is_active or not current_user.is_authenticated:
            return False

        if current_user.has_role(self.roles_required):
            return True

        return False

def action_activate(self, ids):
        """Batch registraion of users."""
        try:
            count = self.model.update(is_active=True).where(
                self.model.is_active == False,  # noqa: E712
                self.model.id.in_(ids)).execute()
        except Exception as ex:
            flash(f"Failed to activate the selected records: {ex}")
            app.logger.exception("Failed to activate the selected records")
        else:
            flash(f"{count} records were activated for batch processing.")

def check(self):
        # Do not override DENY_ABORT_HTTP_CODE because inherited classes will
        # better use HTTP 403/Forbidden code on denial.
        self.DENY_ABORT_HTTP_CODE = HTTPStatus.UNAUTHORIZED
        # NOTE: `is_active` implies `is_authenticated`.
        return current_user.is_active

def login_notrequired(func):
    @wraps(func)
    def wrap(*args, **kwargs):
        if not current_user.is_active:
            return func(*args, **kwargs)
        else:
            return redirect(url_for("regular.home"))
    return wrap

def is_accessible(self):
        if not current_user.is_active or not current_user.is_authenticated:
            return False
        if current_user.is_admin or current_user.is_teacher:
            return True
        return False

def is_active(self):
        return self.active

    # ---------- END FLASK-LOGIN REQUIREMENTS ------------------------------

def newclub_submit():
    '''Upload excel file to create new clubs'''
    if not siteconfig.get_config('allow_club_creation'):
        abort(403)
    clubname = request.form['clubname']
    email = request.form['email']
    clubtype = int(request.form['clubtype'])
    location = request.form['location']
    intro = request.form['intro']
    description = request.form['description'].strip()
    true_or_fail(clubname, 'Please input club name.', 'newclub')
    true_or_fail(email, 'Please input teacher\'s email address.', 'newclub')
    true_or_fail(location, 'Please input club\'s meeting location.', 'newclub')
    true_or_fail(intro, 'Please input club\'s one-sentence introduction.',
                        'newclub')
    true_or_fail(len(intro) <= 90, 'Your one sentence intro is too long.',
                                   'newclub')

    true_or_fail(description, 'Please input club\'s paragraph description.',
                              'newclub')
    if form_is_valid():
        c = Club.new()
        c.name = clubname
        teacher = User.find_teacher(email)
        if teacher is not None:
            c.teacher = teacher
        else:
            fail('There is no teacher with this email address.', 'newclub')
            return redirect(url_for('.newclub'))
        c.leader = current_user
        c.description = FormattedText.emptytext()
        c.location = location
        c.is_active = False
        c.intro = intro
        c.picture = Upload(-101)
        c.type = ClubType(clubtype)
        c.joinmode = ClubJoinMode.FREE_JOIN
        c.reactivate = True
        c.create()
        c.add_member(current_user)
        c.description = FormattedText.handle(current_user, c,
                                             request.form['description'])
        return redirect(url_for('.clubintro', club=c.callsign))
    return redirect(url_for('.newclub'))