🖋️

Django Ninjaで認証機能を追加

2024/04/01に公開

前回まではDjango Ninjaで基本的なCRUDを実装してきましたが、今回は認証機能を実装していきます。

※前回までの記事はこちら※
https://zenn.dev/keita_f/articles/c8ddbe7fe10ef8
https://zenn.dev/keita_f/articles/e2e58903d954e8

スキーマの設定

schemaは前回までに作成したCustomUserモデルをベースに作成し、後から利用するTokenに関するschemaも設定しておきます。

CustomUserの設定はこちらの記事を参照してください。
https://zenn.dev/keita_f/articles/6ce5cc265d3f1f

コードはこちら
backend/accounts/schemas.py
from .models import CustomUser
from ninja import ModelSchema, Schema

class CustomUserSchema(ModelSchema):
  class Meta:
    model = CustomUser
    exclude = ['password', 'last_login', 'birth_date']

class CustomUserOut(ModelSchema):
  class Meta:
    model = CustomUser
    fields = ['uuid', 'username', 'email', 'first_name', 'last_name']

class CreateCustomUser(Schema):
  username: str
  email: str
  password: str
  first_name: Optional[str]
  last_name: Optional[str]

class UpdateUser(Schema):
  username: Optional[str]
  email: Optional[str]
  first_name: Optional[str]
  last_name: Optional[str]
  is_active: Optional[bool]
  is_staff: Optional[bool]
  is_superuser: Optional[bool]

# token
class TokenSchema(Schema):
  access_token: str
  refresh_token: str
  token_type: str

CustomUserSchemaは基本的なschemaとして設定しています。
そしてDjango Ninjaの最大のメリットである、Djangoモデルからschemaを生成をするため、ModelSchemaクラスを渡して、schemaを作成しています。

詳細は公式ドキュメントを参照してください。
https://django-ninja.dev/guides/response/django-pydantic/

ModelSchemaクラスを渡すことで、Djangoで設定したモデルとそのフィールドを利用してschemaを生成できます。
また、今回のCustomUserSchemaは、DjangoのモデルであるCustomUserを使用しているため、json形式で出力するときに含めたくないフィールドも含まれています。
そのため、json形式に含めたくない(=外部に出したくないフィールド)をexcludeに配列で指定し、json形式での出力からは除いています。
このようにすることで、不要なフィールドや外部に出したくないフィールドを制限することができます。

また、CustomUserOutのschemaでは、同じようにModelSchemaを渡して、Djangoのモデルからschemaを生成していますが、fieldsを指定し、json形式で返したいフィールドだけを指定しています。
ここでは、5つのフィールドのみを指定しています。

Tokenスキーマについては、この後使用するために3つのフィールドを設定しています。
これはDjnagoのモデルを使用しませんので、Schemaクラスを渡しています。

認証のための処理

次に、認証を行うための処理を書いていきます。
Django Ninjaでの認証は、以下の公式ドキュメントとChatGPTを参考に設定しました。
必ずしも正解というわけではありませんので、適宜最適な方法で実装してください。
https://django-ninja.dev/guides/authentication/

それでは、認証に必要なクラスを設定していきます。

コードはこちら
backend/accounts/crud.py
from .models import CustomUser
from ninja.security import HttpBearer
from jose import jwt

class JWTBearer(HttpBearer):
  def authenticate(self, request, token):
    try:
      # tokenを検証してペイロードを取得
      payload = jwt.decode(
        token,
        os.environ['SECRET_KEY'],
        algorithms = os.environ['ALGORITHM']
      )
      # uuidを取得してuserを取得
      user = CustomUser.objects.get(uuid = payload['uuid'])
      return user
    except Exception as e:
      return {
        'status_code': 401,
        'detail': 'トークンタイプが一致していません'
      }

公式ドキュメントにあるHTTP Bearerを使用して設定しています。
また、python-joseを使用したjwtの処理を追記し、渡されたtokenをdecodeして、ユーザー取得のためuuidを取り出しています。

そして、このように設定したクラスをAPI作成時に、各APIにauth引数として渡すことで設定したAPIに認証を必須とすることができるようになります。
また、Routerにauth引数を渡すことで、routerに紐づくすべてのAPIに認証機能を設定することができます。

from .crud import JWTBearer
router = Router(auth = JWTBearer())

認証を設定したAPIの例
(上記の例はauthorに関するAPIに認証を必須としたものです)

tokenの生成

認証にはjwtを使用したtoken認証としていますので、これに必要なaccess_tokenとrefresh_tokenを作成していきます。

コードはこちら
backend/accounts/crud.py
from datetime import datetime, timedelta

from .models import CustomUser

from ninja.security import HttpBearer

from jose import jwt

import os
from dotenv import load_dotenv # add

load_dotenv()

def create_access_token(user):
  # access_tokenの有効期限を3時間に設定
  expire = datetime.now() + timedelta(hours = 3)
  return encode_token(user, expire)

def create_refresh_token(user):
  # refresh_tokenの有効期限を1週間に設定
  expire = datetime.now() + timedelta(weeks = 1)
  return encode_token(user, expire)

def encode_token(user, expire):
  # JWTトークンのpayloadを設定
  payload = {
    'uuid':user.uuid,
    'exp': expire
  }
  
  # JWTトークンを生成して返す
  return jwt.encode(
    payload,
    os.environ['SECRET_KEY'],
    algorithm = os.environ['ALGORITHM']
  )

access_tokenの生成:create_access_token
refresh_tokenの生成:create_refresh_token
JWTtokenの生成:encode_token
としています。

APIの処理はこの後になりますが、ログインAPIを叩いた時に、userを取得することにするので、そのuserをこれらの関数に渡し、userからuuidを取り出し、tokenの生成に使用しています。
また、SECRET_KEYとALGORITHMは、使い回しができるようにかつ外部に漏れないように、.envファイルに記載しています。
なお、SECRET_KEYの生成は、以下のコマンドを実行します。

openssl rand -hex 32  

これで生成されたSECRET_KEYを.envファイルに記録します。

APIの作成

それではAPIを作成していきます。
今回は、authorアプリケーション内に作成したapi.pyに書いていきAPIに原則認証が必要となるような設定にしていきます。

コードはこちら
backend/accounts/api.py
from typing import List

from .models import CustomUser
from django.shortcuts import get_object_or_404
from django.contrib.auth import get_user

from ninja import Router
from .schemas import CustomUserOut, TokenSchema

from .crud import create_access_token, create_refresh_token, JWTBearer

router = Router(auth = JWTBearer())

@router.get('/author/{uuid}', response = CustomUserOut)
def get_author(request, uuid: str):
  author = get_object_or_404(CustomUser, uuid = uuid)
  return author

@router.get('/authors', response = List[CustomUserOut], auth = JWTBearer())
def get_authors(request):
  authors = CustomUser.objects.all()
  return authors

@router.post('/auth/login', response = TokenSchema, auth = None)
def login(request, email: str, password: str):
  user = CustomUser.objects.get(email = email)
  if user.check_password(password):
    access_token = create_access_token(user)
    refresh_token = create_refresh_token(user)
    return {
      'access_token': access_token,
      'refresh_token': refresh_token,
      'token_type': 'bearer'
    }
  else:
    return {
      'detail': 'Incorrect email or password'
    }

@router.get('/auth/get_login_user', auth = JWTBearer())
def get_login_user(request):
  login_user = get_user(request)
  if login_user.is_authenticated:
    return login_user
  else:
    return {
      'detail': 'User not authenticated.'
    }

ログインAPI

@router.post('/auth/login', response = TokenSchema, auth = None)
def login(request, email: str, password: str):
  user = CustomUser.objects.get(email = email)
  if user.check_password(password):
    access_token = create_access_token(user)
    refresh_token = create_refresh_token(user)
    return {
      'access_token': access_token,
      'refresh_token': refresh_token,
      'token_type': 'bearer'
    }
  else:
    return {
      'detail': 'Incorrect email or password'
    }

ログイン用のAPIではresponseとしてtokenを返す(取得する)ようにしますので、response引数にTokenSchemaを指定しています。
また、上記のとおり、ログイン用のAPIに関しては認証を不要とするので、auth引数にはNoneを指定し、認証を適用していません。
一方、routerではauthに先ほど書いたJWTBearerを指定していますので、ログイン用のAPI以外には、個別にauth引数を指定しなくても、基本的に認証が必要な設定となります。

処理の流れとしては、

  1. emailとpasswordを受け取りログイン処理をする
  2. ログインはemailに紐づくデータをCustomUserモデルから取得している
  3. userのpasswordをチェック(Djangoにデフォルトであるcheck_passwordを使用)し、問題なければ、access_tokenとrefresh_token、token_typeを返す
  4. passwordのチェックが通らなければ、emailまたはpasswordに誤りがあることを通知してる
    という流れです。

authorの取得

authorの取得は、6種類用意しました。
(R6.4.1:post, put, deleteメソッドを追加しました)

コードはこちら
@router.get('/author/{uuid}', response = CustomUserOut)
def get_author(request, uuid: str):
  author = get_object_or_404(CustomUser, uuid = uuid)
  return author

@router.get('/authors', response = List[CustomUserOut])
def get_authors(request):
  authors = CustomUser.objects.all()
  return authors

@router.get('/auth/get_login_user', response = CustomUserOut)
def get_login_user(request: HttpRequest):
  login_user = request.auth
  if login_user.is_authenticated:
    return login_user
  else:
    return {
      'detail': 'User not authenticated.'
    }

@router.post('/create', response = CustomUserOut, auth = None)
def create_user(request, payload: CreateCustomUser):
  if CustomUser.objects.filter(email = payload.email).exists():
    return {
      'message': 'Email is already used. Please change email address.'
    }
  if CustomUser.objects.filter(username = payload.username).exists():
    return {
      'message': 'Username is already is used. Please other username set.'
    }
  
  user = CustomUser(
    uuid = str(uuid.uuid4()) # uuidはデフォルトでuuid型になるためstrに変換
    username = payload.username,
    email = payload.email,
    first_name = payload.first_name,
    last_name = payload.last_name
  ) 
  user.set_password(payload.password)
  user.save()
  return user

@router.put('/author/{uuid}', response = CustomUserOut)
def update_author(request, uuid: str, payload: UpdateUser):
  author = get_object_or_404(CustomUser, uuid = uuid)
  for attr, value in payload.dict().items():
    setattr(author, attr, value)
  author.save()
  return author

@router.delete('/author/{uuid}')
def delete_author(request, uuid: str):
  author = get_object_or_404(CustomUser, uuid = uuid)
  author.delete()
  return {'message': 'success'}

いずれもGETメソッドなので特に難しいことはしていません。
認証が通ればauthorに関するデータを取得できるようにしています。
また、最後のget_login_userは、request.authを使用してログイン中のユーザーを取得しています(認証の確認はDjangoのis_authenticatedを使用)。
post, put, deleteはarticleで使用したAPIを流用しています。

このような画面になればOKです(画像は認証後のものです)。

最後に

Django Ninjaでログイン処理を実装してみました。
まだ、refresh_tokenを使用していないことと、authorの更新や削除などのauthorに関するCRUD処理を設定できていないのがまだまだといったところです(今後この記事を更新していく予定です)。

CRUD処理を作ったならば、その後はフロントエンド(Next.jsかSveltkitまたはhonoxを検討中)から実際に情報を送信して、データの取得ができるかも試してみたいと思っています。

Discussion