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

def login():
    if request.method == 'POST':
        if request.form['user'] is '' or request.form['password'] is '':
            flash("Can't leave it blank",'danger')
        else:
            au=auth()
            user=au.check(request.form['user'],request.form['password'])
            if user:
                login_user(user)
                flash('Logged in successfully.','success')
                if user.is_admin:
                        return redirect(url_for('admin'))
                return redirect(url_for('desktops'))
            else:
                flash('Username not found or incorrect password.','warning')
    remote_addr=request.headers['X-Forwarded-For'] if 'X-Forwarded-For' in request.headers else request.remote_addr
    disposables=app.isardapi.show_disposable(remote_addr)
    log.info(disposables)
    log.info(remote_addr)
    return render_template('login_disposables.html', disposables=disposables if disposables else '')

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

    session = Session()
    username = request.json.get('username', '')
    password = request.json.get('password', '')

    registered_user = (session.query(User)
                            .filter(and_(User.username==username)).first())

    if (registered_user is None or not
            check_password_hash(registered_user.password, password)):
        Session.remove()
        return jsonify(success=False)

    login_user(registered_user)
    flash('Logged in successfully')
    Session.remove()

    user = current_user.__dict__.copy()
    user.pop('_sa_instance_state', None)
    return jsonify(success=True, isAdmin=current_user.is_admin(), **user)

def password_reset(username=None):
    if not current_user.is_admin:
        return redirect("/")
    if username is not None:
        if current_user.id != 'superadmin':
            check_user = db.DJUser.get(username)
            if check_user.is_admin and check_user.id != current_user.id:
                return 'You can\'t reset another admin\'s password.'
        new_pass = db.change_password(username)
        if new_pass is not None:
            return '{}\'s new password is "{}".'.format(username, new_pass)
    return '{} doesn\'t exist.'.format(username)


###############################################################################
# Utility functions

def checkValid(hid, pid):
    problem = Problem.query.filter_by(id = pid).first()
    homework = None
    home_list = problem.homework
    for home in home_list:
        if int(home.id) == hid:
            homework = home
            if not homework.camp.public and not current_user.is_admin :
                valid = False
                for user in home.camp.user:
                    if user.id == current_user.id:
                        valid = True
                        break
                if not valid:
                    flash('????????????')
                    return redirect(request.args.get('next') or url_for("main.index")), False, problem, homework
            break
    return None, True, problem, homework

def edit_profile(username):
    user = User.query.filter(User.username == username).first()
    if not user.id == current_user.id and not current_user.is_admin:
        flash('Insufficient priviliges.')
        return redirect(url_for('users.profile_page', username=username))

    form = ProfileForm(request.form)
    if request.method == 'POST' and form.validate():
        user.description = form.bio.data
        db_session.add(user)
        db_session.commit()
        flash('Profile saved.')
        return redirect(url_for('users.edit_profile', username=user.username))

    return render_template('users/edit_profile.html', user=user, form=form)

def admin_required(func):
    """Check that user is logged in and an administrator."""
    @wraps(func)
    def decorated_view(*args, **kwargs):
        # See implementation of flask_login.utils.login_required
        if request.method in EXEMPT_METHODS:
            return func(*args, **kwargs)
        elif login_manager._login_disabled:
            return func(*args, **kwargs)
        elif not (current_user.is_authenticated and current_user.is_admin):
            return login_manager.unauthorized()
        return func(*args, **kwargs)
    return decorated_view

def admin_required(func):
    @functools.wraps(func)
    def _admin_required(*args, **kw):
        admin = current_user.is_authenticated and current_user.is_admin
        if not admin:
            return login_manager.unauthorized()
        return func(*args, **kw)
    return _admin_required

def admin_required(f):
    @wraps(f)
    def decorated_function(*args, **kwargs):
        if not current_user.is_admin():
            return "You do not have access", 403 
        return f(*args, **kwargs)
    return decorated_function

def template_helpers():
    def guess_tab(path):
        if path.startswith("/geno"):
            return "geno"
        elif path.startswith("/pheno"):
            return "pheno"
        elif path.startswith("/jobs") or path == "/":
            return "job"
        elif path == "/admin":
            return "job"
        elif path.startswith("/admin/user"):
            return "user"
        elif path.startswith("/admin/phenos"):
            return "pheno"
        else:
            return ""

    def get_navigation_links(path, user=None):
        links = {"left": [], "right":[]}
        if path.startswith("/admin"):
            links["left"].append(("job", "Jobs", url_for("get_admin_page")))
            links["left"].append(("user", "Users", url_for("get_admin_user_page")))
            links["left"].append(("pheno", "Phenos", url_for("get_admin_pheno_page")))
            links["right"].append(("return","Return to App", url_for("index")))
        else:
            links["left"].append(("job", "Jobs", url_for("index")))
            links["left"].append(("pheno", "Phenotypes", url_for("get_pheno_list")))
            if (user is not None) and hasattr(user, "is_admin") and user.is_admin():
                links["right"].append(("admin","Admin", url_for("get_admin_page")))
        links["right"].append(("logout","Logout", url_for("sign_out")))
        return links

    return dict(guess_tab = guess_tab, 
        get_navigation_links = get_navigation_links)


# @app.errorhandler(500)
# def internal_error(exception):
#     return render_template('500.html'), 500

def index():
    try:
        if current_user.is_authenticated:
            if current_user.is_admin:
               return redirect(url_for('admin'))
        else:
            title='Sign in to start'
    except Exception as e:
        print("Something went wrong with username? Exception:",e)
    remote_addr=request.headers['X-Forwarded-For'] if 'X-Forwarded-For' in request.headers else request.remote_addr
    disposables=app.isardapi.show_disposable(remote_addr)
    log.info(disposables)
    log.info(remote_addr)    
    return render_template('login_disposables.html', disposables=disposables if disposables else '')

def isAdmin(fn):
    @wraps(fn)
    def decorated_view(*args, **kwargs):
        if current_user.is_admin:
            return fn(*args, **kwargs)
        logout_user()
        return redirect(url_for('index'))
    return decorated_view

def logged_in():
    try:
        is_admin = current_user.is_admin()
    except Exception:
        is_admin = false
    return jsonify(success=True, 
                   loggedIn=bool(current_user.is_authenticated),
                   isAdmin=bool(is_admin))

def is_accessible(self):
        return current_user.is_admin()

def admin_dashboard():
    # prevent non-admins from accessing the page
    if not current_user.is_admin:
        abort(403)

    return render_template('home/admin_dashboard.html', title="Dashboard")

def check_admin():
    # prevent non-admins from accessing the page
    if not current_user.is_admin:
        abort(403)


# Department Views

def assign_employee(id):
    """
    Assign a department and a role to an employee
    """
    check_admin()

    employee = Employee.query.get_or_404(id)

    # prevent admin from being assigned a department or role
    if employee.is_admin:
        abort(403)

    form = EmployeeAssignForm(obj=employee)
    if form.validate_on_submit():
        employee.department = form.department.data
        employee.role = form.role.data
        db.session.add(employee)
        db.session.commit()
        flash('You have successfully assigned a department and role.')

        # redirect to the roles page
        return redirect(url_for('admin.list_employees'))

    return render_template('admin/employees/employee.html',
                           employee=employee, form=form,
                           title='Assign Employee')

def streaming_over_page():
    if current_app.config['livestream']['current_dj'] == current_user.id or \
       current_user.is_admin:
        current_app.config['livestream']['active'] = False
        current_app.config['livestream']['last_played'] = []
        current_app.config['livestream']['current_dj'] = None
        return 'Switched back to regular programming.'
    return 'You are not streaming!'

def admin_page():
    if not current_user.is_admin:
        return redirect("/")
    if request.method == 'GET':
        user_status = db.get_a_list(['banned', 'admin'])
        return render_template("admin.html", all_users=user_status)
    else:
        username = request.form['username']
        new_user = db.make_user(username)
        if new_user is None:
            return '{} already exists!'.format(username)
        else:
            return '{} created, with password "{}".'.format(*new_user)

def ban_user(username=None):
    if not current_user.is_admin:
        return redirect("/")
    if username is not None:
        if username == current_user.id:
            return 'Don\'t ban yourself.'
        check_user = db.DJUser.get(username)
        if check_user.is_admin:
            return 'You can\'t ban an admin.'
        ban_result = db.update_dj_status(username, 'banned', True)
        if ban_result is not None:
            return '{} is now banned.'.format(username)
    return '{} doesn\'t exist.'.format(username)

def unban_user(username=None):
    if not current_user.is_admin:
        return redirect("/")
    if username is not None:
        ban_result = db.update_dj_status(username, 'banned', False)
        if ban_result is not None:
            return '{} is now unbanned.'.format(username)
    return '{} doesn\'t exist.'.format(username)

def admin_required(func):
    @wraps(func)
    def decorated_view(*args, **kwargs):
        if not current_user.is_admin:
            return app.login_manager.unauthorized()
        return func(*args, **kwargs)
    return decorated_view

def dashboard():
    """
    not admin is the one can only reach that page, everone does
    """
    if current_user.is_admin:
        abort(403)

    return render_template('home/dashboard.html', title="Dashboard")

def admin_dashboard():
    # prevent non-admins from accessing the page
    if not current_user.is_admin:
        abort(403)

    return render_template('home/admin_dashboard.html', title="Dashboard")

def check_admin():
    """
    Prevent non-admins from accessing the page
    """
    if not current_user.is_admin:
        abort(403)

# Category Views

def kitap_kirala(id):
    """
    Shows the list of the books can be rented for the users, not admins.
    It will pop N/A for the lovely admins
    """
    check_admin()

    uye = Uye.query.get_or_404(id)

    # prevent admin from being renting a book
    if uye.is_admin:
        abort(403)

    form = UyeForm(obj=uye)
    if form.validate_on_submit():
        yeni_kitap = form.kitap.data
        yeni_kitap.uye = Uye.query.filter_by(id=id).first()
        db.session.add(yeni_kitap)
        db.session.commit()
        flash('Kiralama islemi gerceklesmistir.')
        # redirect to the renting page
        return redirect(url_for('admin.kiralama_listesi'))

    return render_template('admin/kiralama/kirala.html',
                           uye=uye, form=form,
                           title='Kitap Kirala')

def edit_reports():
    # Get the fields from the add-new-filters form
    user_id = None
    if not current_user.is_admin:
        user_id=current_user.user_id
    return_data = get_reports_data(False, user_id)
    return render_template(
        'public/reports_management.html',
        report_data=return_data,
        report_meta_fields=get_report_metadata_fields(),
        api_token=current_user.api_token)

def admin_panel():
    form = AdminForm()
    if not current_user.is_admin:
        abort(403)
    else:
        users_data = db.session.query(User).all()
        return render_template('users/admin.html', users_data=users_data, form=form)

def admin_dashboard():
    """
    Render the dashboard for admin
    :return: template home/admin_dashboard.html
    """
    # prevent non-admins from accessing the page
    if not current_user.is_admin:
        abort(403)

    return render_template('home/admin_dashboard.html',
                           title="Dashboard")

def check_admin():
    """
    Prevent non-admins from acessing the page
    :return: 403 - forbidden page
    """
    if not current_user.is_admin:
        abort(403)

def assign_employee(id):
    """
    Assign a department and a role to an employee
    """
    check_admin()

    employee = Employee.query.get_or_404(id)

    # prevent admin from being assigned a department or role
    if employee.is_admin:
        abort(403)

    form = EmployeeAssignForm(obj=employee)
    if form.validate_on_submit():
        employee.department = form.department.data
        employee.role = form.role.data
        db.session.add(employee)
        db.session.commit()
        flash('You have successfully assigned a department and role.')

        # redirect to the roles page
        return redirect(url_for('admin.list_employees'))

    return render_template('admin/employees/employee.html',
                           employee=employee, form=form,
                           title='Assign Employee')

def admin_dashboard():
    # prevent non-admins from accessing the page
    if not current_user.is_admin:
        abort(403)

    return render_template('home/admin_dashboard.html', title="Dashboard")

def check_admin():
    """
    Prevent non-admins from accessing the page
    """
    if not current_user.is_admin:
        abort(403)

# Department Views

def assign_employee(id):
    """
    Assign a department and a role to an employee
    """
    check_admin()

    employee = Employee.query.get_or_404(id)

    # prevent admin from being assigned a department or role
    if employee.is_admin:
        abort(403)

    form = EmployeeAssignForm(obj=employee)
    if form.validate_on_submit():
        employee.department = form.department.data
        employee.role = form.role.data
        db.session.add(employee)
        db.session.commit()
        flash('You have successfully assigned a department and role.')

        # redirect to the roles page
        return redirect(url_for('admin.list_employees'))

    return render_template('admin/employees/employee.html',
                           employee=employee, form=form,
                           title='Assign Employee')

def admin_required(func):
    @wraps(func)
    def wrap(*args, **kwargs):
        if current_user.is_admin():
            return func(*args, **kwargs)
        else:
            flash("Error accessing page - admin priviledges needed")
            return redirect(url_for("regular.home"))
    return wrap

def code_show(sid):
    print('come!')
    sub = Submission.query.filter_by(id = sid).first()
    if sub is not None:
        if (not current_user.is_admin) and (not current_user.is_teacher) and current_user.id != sub.user.id:
            flash('????????????')
            return redirect('/status')
        p_list = []
        id = 0
        print('source type', sub.source)
        if sub.source[-2:] == 'py':
            print('first')
            filename ='source.py'
            path = os.path.join(app.config['UPLOAD_FOLDER'], 'submission', sid, filename)
            fd = open(path, 'r')
            content = fd.read()
            content = content.strip(' \t')
            return render_template('code_view.html', code_list = [['source.py', '1', content]], user = sub.user, prob = sub.prob, sid=sid)
        else:
            print('second')
            for parent, dir, filenames in os.walk(os.path.join(app.config['UPLOAD_FOLDER'], 'submission', sid)):
                print('parent', parent)
                print('dir', dir)
                for filename in filenames:
                    print('filename: ', filename)
                    if filename[-3:] == '.py':
                        path = os.path.join(parent, filename)
                        print('path: ', path)
                        fd = open(path, 'r')
                        content = fd.read()
                        id+=1
                        p_list.append([filename, str(id), content])
            return render_template('code_view.html', code_list = p_list, user = sub.user, prob = sub.prob, sid = sid)
    flash('????????!')
    return redirect("/admin/submission/")

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_accessible(self):
        if AdminView.is_accessible(self) and current_user.is_admin:
            return True
        return False

def is_accessible(self):
        if AdminView.is_accessible(self) and current_user.is_admin:
            return True
        return False

def edit_view(self):
        id = request.args.getlist('id')[0]
        if id is None:
            return redirect('/admin')
        model = self.get_one(id)
        if not current_user.is_admin and model.owner.id != current_user.id:
            flash('????????')
            return redirect('/admin')
        return AdminView.edit_view(self)

def on_model_delete(self, model):
        if current_user.is_admin or (current_user.is_teacher and current_user.id == model.owner.id):
            return
        raise ValidationError("??????????????")

def edit_view(self):
        id = request.args.getlist('id')[0]
        if id is None:
            return redirect('/admin')
        model = self.get_one(id)
        if not current_user.is_admin and model.owner.id != current_user.id:
            flash('????????')
            return redirect('/admin')
        return AdminView.edit_view(self)

def on_model_delete(self, model):
        if current_user.is_admin or (current_user.is_teacher and current_user.id == model.owner.id):
            return
        raise ValidationError("??????????????")

def edit_view(self):
        id = request.args.getlist('id')[0]
        if id is None:
            return redirect('/admin')
        model = self.get_one(id)
        if not current_user.is_admin and model.author.id != current_user.id:
            flash('????????')
            return redirect('/admin')
        return AdminView.edit_view(self)

def on_model_delete(self, model):
        if current_user.is_admin or (current_user.is_teacher and current_user.id == model.author.id):
            return
        raise ValidationError("??????????????")

def on_model_change(self, form, model, is_created):
        if current_user.is_admin or (current_user.is_teacher and current_user.id == model.owner.id):
            return
        if is_created:
            raise ValidationError("?????????????????")
        else:
            raise ValidationError("?????????????????")

def on_model_delete(self, model):
        if current_user.is_admin or (current_user.is_teacher and current_user.id == model.owner.id):
            return
        raise ValidationError("???????????????")

def get_query(self):
        if current_user.is_admin:
            return Submission.query
        elif current_user.is_teacher:
            return Submission.query.join(HomeWork).join(TrainCamp).filter(or_(HomeWork.owner_id == current_user.id, TrainCamp.public == True))
        else:
            return None

def is_accessible(self):
        if AdminView.is_accessible(self) and current_user.is_admin:
            return True
        return False

def teacher_required(view_func):
    @wraps(view_func)
    def decorator(*args, **kwargs):
        if current_user.is_teacher or current_user.is_admin:
            return view_func(*args, **kwargs)
        else:
            flash("????????????")
            return redirect(url_for('main.login', next = request.url))
    return decorator

def admin_dashboard():
    # prevent non-admins from accessing the page
    if not current_user.is_admin:
        abort(403)

    return render_template('home/admin_dashboard.html', title="Dashboard")