flas中的键错误

2024-04-23 23:19:49 发布

您现在位置:Python中文网/ 问答频道 /正文

我在使用烧瓶创建的浏览器中打开页面时出错。在

我认为错误在于使用get_缓存。我试图解决这个错误,但无法得到确切的答案。我用flask_caching import导入了缓存缓存。那个我得到错误的主视图代码如下:

class InvalidUser(Exception):
    pass

class InvalidAction(Exception):
    pass

@main_views.route('/')
def index():
    return render_template("index.html")

class BaseQuestionView(MethodView):
    decorators = [auth_required('session')]

    def get_model(self, id):
        question = Question.query.filter_by(id=id).first()
        if question.user == current_user:
            return question
        else:
            raise InvalidUser()

    def get_cache_key(self, id):
        return "{0}_question_status".format(id)

    def get_cache(self, id):
        from app import cache
        return cache.get(self.get_cache_key(id))

    def set_cache(self, id, val):
        from app import cache
        return cache.set(self.get_cache_key(id), val)

class QuestionView(BaseQuestionView):
    def get_questions(self):
        questions = current_user.questions
        for q in questions:
            q.status = self.get_cache(q.id)
        return questions

    def get(self):
        form_obj = QuestionForm()
        form = render_template("forms/question.html", form=form_obj)
        return render_template("questions.html", questions=self.get_questions(), form=form)


    def post(self):
        form_obj = QuestionForm()
        if form_obj.validate_on_submit():
            form_obj.save()
            return redirect(url_for('main_views.questions'))
        form = render_template("forms/question.html", form=form_obj)

        return render_template("questions.html", questions=self.get_questions(), form=form)

main_views.add_url_rule('/questions', view_func=QuestionView.as_view('questions'))

class QuestionDetailView(BaseQuestionView):

    def get_upload_form(self, model):
        upload_obj = EssayUploadForm()
        upload_form = render_template("forms/upload.html", form=upload_obj, question=model)
        return upload_form

    def get(self, id):
        model = self.get_model(id)
        form_obj = EssayForm()

        form = render_template("forms/essay.html", form=form_obj, question=model)
        return render_template("question.html", question=model, form=form, status=self.get_cache(id), upload_form=self.get_upload_form(model))

    def post(self, id):
        form_obj = EssayForm()
        model = self.get_model(id)
        if form_obj.validate_on_submit():
            form_obj.save(model)
            return redirect(url_for('main_views.question_detail', id=id))
        form = render_template("forms/essay.html", form=form_obj, question=model)
        return render_template("question.html", question=model, form=form, status=self.get_cache(id), upload_form=self.get_upload_form(model))

    def delete(self, id):
        from app import db
        question = self.get_model(id)
        db.session.delete(question)
        db.session.commit()
        return jsonify({})

main_views.add_url_rule('/question/<int:id>', view_func=QuestionDetailView.as_view('question_detail'))

class QuestionActionView(BaseQuestionView):
    actions = ["create", "create_and_score", ]

    def create(self, model):
        from core.tasks.tasks import create_model
        task_status = create_model.delay(model.id)
        return task_status.id

    def create_and_score(self, model):
        from core.tasks.tasks import create_and_score
        task_status = create_and_score.delay(model.id)
        return task_status.id

    def get(self, id, action):
        model = self.get_model(id)
        if action not in self.actions:
            raise InvalidAction()
        task_id = getattr(self, action)(model)
        self.set_cache(id, url_for('main_views.task_status', task_id=task_id))
        return jsonify({})

main_views.add_url_rule('/question/<int:id>/<string:action>', view_func=QuestionActionView.as_view('question_action'))

class EssayUploadView(BaseQuestionView):

    def post(self, id):
        form_obj = EssayUploadForm()
        if form_obj.validate_on_submit():
            model = self.get_model(id)
            file = request.files['upload']
            form_obj.save(model, file)

        return redirect(url_for('main_views.question_detail', id=id))

main_views.add_url_rule('/question/<int:id>/upload_essays', view_func=EssayUploadView.as_view('essay_upload'))

class BaseEssayView(MethodView):
    decorators = [auth_required('session')]

    def get_model(self, id):
        essay = Essay.query.filter_by(id=id).first()
        if essay.question.user == current_user:
            return essay
        else:
            raise InvalidUser()

    def get_cache_key(self, id):
        return "{0}_essay_status".format(id)

    def get_cache(self, id):
        from app import cache
        return cache.get(self.get_cache_key(id))

    def set_cache(self, id, val):
        from app import cache
        return cache.set(self.get_cache_key(id), val)

class EssayDetailView(BaseEssayView):

    def delete(self, id):
        from app import db
        essay = self.get_model(id)
        db.session.delete(essay)
        db.session.commit()
        return jsonify({})

main_views.add_url_rule('/essay/<int:id>', view_func=EssayDetailView.as_view('essay_detail'))

class EssayActionView(BaseEssayView):
    actions = ["score", ]

    def score(self, model):
        from core.tasks.tasks import score_essay
        task_status = score_essay.delay(model.id)
        return task_status.id

    def get(self, id, action):
        model = self.get_model(id)
        if action not in self.actions:
            raise InvalidAction()
        task_id = getattr(self, action)(model)
        self.set_cache(id, url_for('main_views.task_status', task_id=task_id))
        return jsonify({})

main_views.add_url_rule('/essay/<int:id>/<string:action>', view_func=EssayActionView.as_view('essay_action'))

class TaskStatusView(MethodView):
    decorators = [auth_required('session')]

    def get(self, task_id):
        from app import celery

        result = celery.AsyncResult(task_id)
        state, retval = result.state, result.result
        response_data = dict(id=task_id, status=state)
        if state in states.EXCEPTION_STATES:
            traceback = result.traceback
            response_data.update({'exc': get_full_cls_name(retval.__class__),
                                  'traceback': traceback})
        return jsonify({'task': response_data})

main_views.add_url_rule('/tasks/<string:task_id>', view_func=TaskStatusView.as_view('task_status'))

回溯:

^{pr2}$

Tags: selfformviewidobjcachetaskget