Implementing JWT Authentication Decorator

Tornado's built-in authenticated decorator relies on session-based authentication. When utilizing JWT, the token must be extracted from the request header, decoded, and used to populate self._current_user. The decorator must also await the decorated coroutine method.

import functools
import jwt
from apps.users.models import User

def require_jwt_auth(method):
    @functools.wraps(method)
    async def wrapper(self, *args, **kwargs):
        auth_token = self.request.headers.get("X-Auth-Token", None)
        if not auth_token:
            self.set_status(401)
            self.finish({})
            return

        try:
            payload = jwt.decode(
                auth_token, 
                self.settings["secret_key"], 
                algorithms=["HS256"], 
                leeway=self.settings["jwt_expire"], 
                options={"verify_exp": True}
            )
            user_id = payload["id"]
            
            try:
                user = await self.application.objects.get(User, id=user_id)
                self._current_user = user
                await method(self, *args, **kwargs)
            except User.DoesNotExist:
                self.set_status(401)
        except jwt.ExpiredSignatureError:
            self.set_status(401)
        
        self.finish({})
        
    return wrapper

The leeway parameter in jwt.decode configures the allowed expiration window, while options={"verify_exp": True} enforces validation of the expiration claim.

Building Group Creation Functionality

To handle group creation with file uploads, avoid JSON forms. Instead, access form data from self.request.body_arguments and uploaded files from self.request.files. Use the aiofiles library for asynchronous file writing.

Model definition:

from datetime import datetime
from peewee import *
from MxForm.models import BaseModel
from apps.users.models import User

class Club(BaseModel):
    owner = ForeignKeyField(User, verbose_name="Owner")
    title = CharField(max_length=100, null=True, verbose_name="Title")
    group_type = CharField(max_length=20, verbose_name="Category", null=True)
    cover_image = CharField(max_length=200, null=True, verbose_name="Cover Image")
    description = TextField(verbose_name="Description")
    announcement = TextField(verbose_name="Announcement")
    member_count = IntegerField(default=0, verbose_name="Members")
    post_count = IntegerField(default=0, verbose_name="Posts")

    @classmethod
    def fetch_with_owner(cls):
        return cls.select(cls, User.id, User.nick_name).join(User)

class ClubMember(BaseModel):
    user = ForeignKeyField(User, verbose_name="User")
    club = ForeignKeyField(Club, verbose_name="Club")
    status = CharField(choices=[("approved", "Approved"), ("rejected", "Rejected")], max_length=10, null=True, verbose_name="Status")
    status_message = CharField(max_length=200, null=True, verbose_name="Status Message")
    request_reason = CharField(max_length=200, verbose_name="Reason")
    processed_at = DateTimeField(default=datetime.now, verbose_name="Processed At")

Form validation:

from wtforms_tornado import Form
from wtforms import StringField, TextAreaField
from wtforms.validators import DataRequired, AnyOf

class ClubCreationForm(Form):
    title = StringField("Title", validators=[DataRequired("Title is required")])
    group_type = StringField("Type", validators=[AnyOf(values=["Education", "Trading", "Development", "Lifestyle"])])
    description = TextAreaField("Description", validators=[DataRequired(message="Description is required")])
    announcement = TextAreaField("Announcement", validators=[DataRequired(message="Announcement is required")])

Request handler:

import os
import uuid
import aiofiles
from playhouse.shortcuts import model_to_dict
from MxForm.handler import *
from apps.utils.mxform_decorators import require_jwt_auth
from apps.community.forms import *
from apps.community.models import *
from apps.utils.util_func import _serial

class ClubHandler(BaseHandler):
    async def get(self, *args, **kwargs):
        pass

    @require_jwt_auth
    async def post(self, *args, **kwargs):
        re_data = {}
        club_form = ClubCreationForm(self.request.body_arguments)
        
        if club_form.validate():
            uploaded_files = self.request.files.get("cover_image", None)
            if not uploaded_files:
                self.set_status(400)
                re_data["cover_image"] = "Cover image is required"
            else:
                saved_filename = ""
                for file_meta in uploaded_files:
                    original_name = file_meta["filename"]
                    saved_filename = f"{uuid.uuid4().hex}_{original_name}"
                    save_path = os.path.join(self.settings["UPLOAD_DIR"], saved_filename)
                    async with aiofiles.open(save_path, 'wb') as output_file:
                        await output_file.write(file_meta['body'])
                
                new_club = await self.application.objects.create(
                    Club,
                    owner=self.current_user, 
                    title=club_form.title.data,
                    group_type=club_form.group_type.data, 
                    description=club_form.description.data,
                    announcement=club_form.announcement.data, 
                    cover_image=saved_filename
                )
                re_data["id"] = new_club.id
        else:
            self.set_status(400)
            for field in club_form.errors:
                re_data[field] = club_form.errors[field][0]
                
        self.write(re_data)

Listing Groups

Peewee requires explicit join assembly when querying tables with foreign key relationships. The fetch_with_owner class method handles this. Additionally, Python's standard .dumps cannot natively serialize datetime objects, necessitating a custom serializer.

Serializer utility:

from datetime import datetime, date

def _time(obj):
    if isinstance(obj, (date, datetime)):
        return obj.isoformat()
    raise TypeError("Object not serializable")

List handler:

class ClubHandler(BaseHandler):
    async def get(self, *args, **kwargs):
        re_data = []
        club_query = Club.fetch_with_owner()

        category = self.get_argument("category", None)
        if category:
            club_query = club_query.filter(Club.group_type == category)

        sort_param = self.get_argument("sort", None)
        if sort_param == "latest":
            club_query = club_query.order_by(Club.add_time.desc())
        elif sort_param == "popular":
            club_query = club_query.order_by(Club.member_count.desc())

        max_items = self.get_argument("limit", None)
        if max_items:
            club_query = club_query.limit(int(max_items))

        clubs = await self.application.objects.execute(club_query)
        for club in clubs:
            club_dict = model_to_dict(club)
            club_dict["cover_image"] = f"{self.settings['SITE_URL']}/media/{club_dict['cover_image']}/"
            re_data.append(club_dict)

        self.finish(.dumps(re_data, default=_serial))

Applying to Join a Group

Form:

class MembershipRequestForm(Form):
    request_reason = StringField("Reason", validators=[DataRequired("Reason is required")])

Handler:

class ClubMembershipHandler(RedisHandler):
    @require_jwt_auth
    async def post(self, club_id, *args, **kwargs):
        re_data = {}
        payload = .loads(self.request.body.decode("utf8"))
        form = MembershipRequestForm.from_(payload)
        
        if form.validate():
            try:
                club = await self.application.objects.get(Club, id=int(club_id))
                await self.application.objects.get(ClubMember, club=club, user=self.current_user)
                self.set_status(400)
                re_data["non_fields"] = "Already a member"
            except Club.DoesNotExist:
                self.set_status(404)
            except ClubMember.DoesNotExist:
                membership = await self.application.objects.create(
                    ClubMember, 
                    club=club,
                    user=self.current_user,
                    request_reason=form.request_reason.data
                )
                re_data["id"] = membership.id
        else:
            self.set_status(400)
            for field in form.errors:
                re_data[field] = form.errors[field][0]
                
        self.finish(re_data)

Group Details

Handler:

class ClubDetailHandler(RedisHandler):
    @require_jwt_auth
    async def get(self, club_id, *args, **kwargs):
        re_data = {}
        try:
            club = await self.application.objects.get(Club, id=int(club_id))
            item_dict = {
                "name": club.title,
                "id": club.id,
                "desc": club.description,
                "notice": club.announcement,
                "member_nums": club.member_count,
                "post_nums": club.post_count,
                "front_image": f"{self.settings['SITE_URL']}/media/{club.cover_image}/"
            }
            re_data = item_dict
        except Club.DoesNotExist:
            self.set_status(404)
            
        self.finish(re_data)

Routing configuration:

from tornado.web import url
from apps.community.handler import *

urlpattern = (
    url("/clubs/", ClubHandler),
    url("/clubs/([0-9]+)/members/", ClubMembershipHandler),
    url("/clubs/([0-9]+)/", ClubDetailHandler),
)