1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

More than 1 year has passed since last update.

Nihon UniversityAdvent Calendar 2023

Day 16

FirebaseAuthの認証をFlaskでやる

Last updated at Posted at 2023-12-16

はじめに

この記事では,PythonでFlaskを使ったREST API(風味)を作ります
また,Firebase Authenticationを使いユーザー認証をさせます

Flaskについてがメインなので,この記事でFirebase関連にはそんなに触れないです

Nihon University Advent Calendar 2023の16日目の記事となっています.ほかの記事もご覧いただければ幸いです

また,似たようなことをやっている記事もあったので,参考に置いておきます

何を作る?

自分用のメモを投下できるアプリを作っていきます.名付けて「$\mathbb{Y}$ memo」です
注:実在するサービス,アプリとは一切関係ないオリジナルのサービスです.

最初にコードみたいって人はは以下のリポジトリを見てください
参考用に段階ごとにブランチを分けてみました.とはいえ途中コミットしたの忘れてて別の段階のやつがはいっちゃったりしてますが.

環境

client側

  • React
  • Vite
  • Firebase
    • Authentication

Bunが速いらしいので使ってみます

(他の方がBunの記事書いてるので見てってください!)

formatterはprettier,linterはeslintです.
Biomeも気になっていますが.Bun + Vite のテンプレートがprettierとeslintだったのでそのままです

server側

  • Flask
  • Firebase Admin SDK

Pythonの環境はRyeを使って管理していきます.理由はRustって流行ってるじゃん?ぐらいです(適当).

Linterはruffを使っていきます.理由はRustって流行って以下略

VSCodeの設定は以下の記事を参考にしました.

では,次からアプリ作っていきます

必要そうな機能

自分用のメモを投下できるアプリにおいて必要そうな機能を考えていきます
(あんまりよくないとは思ってますが見切り発車でやるタイプなので雑です)

client側

  • メモを書く
  • 自分のメモを見る

server側

  • メモを保存する
  • ユーザーごとのメモを渡す

最低限はこんなもんということにしておきます

server側はREST APIとして作るので,そちらも考えます

REST APIへの操作

server側に必要なのは,メモの保存とユーザーの区別でした
なのでそれらに対応するような操作を考えます

REST APIへの操作は以下のようにしてみます

URI GET POST
/users/user_id/memos 特定のユーザーのメモの一覧を取得 メモを追加
/memos 全てのメモの一覧を取得
/memos/memo_id 特定のメモを取得

REST APIの設計と表は以下を参考にしました.

さて,データの保存にはデータベースを使うという認識が私の中にあります.
しかし面倒なので,全てJSON形式のフォルダをローカルに作成されるようにしていきます.

とりあえずメモできるようにする

まずはclient側の基礎を作ります
できたものがこちらです

外観はこちら

image.png

最低限メモができて表示されてるのでOKです
リロードしたらメモしたものはすべて消えるので,メモをserver側に送って保存してもらいたいですね

メモを保存

メモの保存に対応するREST APIへの操作は/users/user_id/memosですが,今は一旦ユーザーについては考えたくないので,/memosに対してのPOSTで保存するようにします
見切り発車の効果が出てきましたかね

基礎を作る

Flaskのインストール

とりあえずFlaskをインストールして作っていきましょう
バージョンが違う方のサイトもあるので注意しましょう

Flaskの使い方

Flaskの基本的な使い方は以下のquickstartを見るのが早いと思います

自分のイメージは

  • Flaskクラスのインスタンスがサーバー本体
  • そのインスタンスのデコレータであるrouteの引数にURLを入れる
    • デコレートする(?)関数が,URLにアクセスしたときの処理

という感じです

サーバーの起動は--appオプションでFlaskクラスのインスタンスを作ったPythonファイルを指定します
flask run --app ファイルパスとコマンドを打てば起動します
ファイル名をapp.pyとすれば--appオプションはいらなくなります1

開発環境では--debugオプションを付けておくと便利です.変更を加えると勝手にリロードされて変更が読み込まれます.
というか逆につけないと変更が適用されず,いちいち起動しなおさないといけなくなります(1敗)

/memosにPOSTする準備

メモを保存するための準備です.

POSTデータの形式

JSON形式で,キーはmemo,値は文字列の配列とします
拡張しやすいようにこうしておきます

JSONファイルを作っておく

memos.jsonという名前にして,dbディレクトリの中に入れました.
このファイルにメモを保存します.

メモを保存する関数の作成

memos.jsonを読み込んでメモを追加して書き込む,という手段でメモを保存します
参考

また,メモの区別としてuuidを生成してidとして入れます

ファイルパスは他の場所でも使いそうだったので定数をまとめる場所を作りました
utilという場所にconstant.pyというファイル作ってそこに定義しておきました

util/constant.py
MEMO_FILE = "./src/db/memos.json"

それと,app.py内に書くと煩雑になりそうな予感がしたので,メモに関する処理をするディレクトリを作ってそこに入れてます(src/memos/save_memo.py

memos.jsonに何も書かれてないのに読み込もうとしたらjson.decoder.JSONDecodeErrorが出たので,そこで空の配列を入れるようにしておきました

書き込みが無事終了したら生成したuuidを,書き込まなかったらNoneを返すようにしています

save_memo.py
import json
import uuid

from util import constant


def save_memo(memo: list[str]) -> bool:
    if not memo or len(memo) == 0:
        return None

    memos = None
    id = str(uuid.uuid4())
    with open(constant.MEMO_FILE, "r") as f:
        try:
            memos = json.load(f)
        except json.decoder.JSONDecodeError:
            memos = []
        memos.append({"memo": memo, "id": id})
    with open(constant.MEMO_FILE, "w") as f:
        json.dump(memos, f, indent=2, ensure_ascii=False)

    return id

POSTされたデータの取得

flaskにrequestというオブジェクトがあり,そこからPOSTされたデータが取得できます.
JSON形式なので,request.get_json()とするとデータが取得可能です.
flaskからimportしましょう

あとは取得したメモをsave_memoに突っ込んで,保存の成否でレスポンスを変えれば出来上がりです.

app.py
from flask import Flask, request
from memos.save_memo import save_memo

app = Flask(__name__)


@app.route("/memos", methods=["POST"])
def index():
    req: dict = request.get_json()
    memo = req.get("memo")
    id = save_memo(memo)
    if id:
        return {"id": id}, 200
    else:
        return {"message": "memo have no content"}, 400

/memosにPOSTしてみる

curlを使えばPOSTできるので試してみましょう.メモの保存ができてると思います

post_memo.sh
curl -X POST -H "Content-Type: application/json" -d '{"memo":"'$1'"}' http://127.0.0.1:5000/memos

client側からPOST

POSTしたらメモが保存されることがわかったので,client側から操作してみましょう.
fetchでURL書いてPOSTを指定すればいいだけと思いきや,http://127.0.0.1:5000だとNo 'Access-Control-Allow-Origin' headerのCORSエラーが発生します.

回避方法はviteのconfigでproxyの設定すればいいらしいんですけどあんまりわかっていません.
ドキュメント読んでください.

とりあえず以下のようにして,/api/memosに対してfetchすればなんとかなります

vite.config.js
import { defineConfig } from "vite";
import react from "@vitejs/plugin-react";

// https://vitejs.dev/config/
export default defineConfig({
  plugins: [react()],
  server: {
    proxy: {
      "/api": {
        target: "http://127.0.0.1:5000",
        changeOrigin: true,
        rewrite: (path) => path.replace(/^\/api/, ""),
      },
    },
  },
});

フォームを送信したときに実行する関数でfetchさせて動作確認しましょう.

// ...前略

  function onMemoSubmit(e) {
    e.preventDefault();
    if (!postMemoRef.current) {
      return;
    }

    const memo_content = postMemoRef.current.value.trim();
    postMemoRef.current.value = "";
    if (memo_content === "") {
      return;
    }

    const parsed_memo_content = memo_content
      .split("\n")
      .filter((line) => line !== "");
    setMemos([...memos, parsed_memo_content]);

    fetch("/api/memos", {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
      },
      body: JSON.stringify({ memo: parsed_memo_content }),
    });
  }
  
// ...後略

以下のような感じになってたらうれしいです

memos.json
[
  {
    "memo": "test",
    "id": "233f8524-9a13-4a3c-9043-a6a4ff69c863"
  },
  {
    "memo": [
      "すももももももももっももももも"
    ],
    "id": "8060f730-3ad4-4ea9-878d-9502e4f9ac6b"
  }
]

メモを取得

続いてメモを取得していきます.

server側

まずは,すべてのメモを取得,およびある一つのメモの取得するための処理を作っていきます.

すべてのメモの取得はJSON読み込めばいいですし,個別のメモもidを参照すれば特定できます.

get_memo.py
import json

from util import constant


def get_all_memos():
    memos = []
    with open(constant.MEMO_FILE, "r") as f:
        try:
            memos = json.load(f)
        except json.decoder.JSONDecodeError:
            memos = []
    return memos


def get_memo(memo_id: str):
    memos = get_all_memos()
    memo = next(filter(lambda m: m["id"] == memo_id, memos), None)
    return memo

ルートの設定

すべてのメモの取得は/memosに対してGET,個別のメモの取得は/memos/memo_idに対してGETでした
なのでこれにそって作っていきます

レスポンスを返す際は,ヘッダーの情報をflaskがいい感じに入れてくれるようにするためにResponseオブジェクトにして返してあげるのがいいんじゃないかなと思います.(ドキュメント

今回使うのは,flaskのjsonify関数です.JSONでの応答してくれます.make_responseでもいいのですが,JSON返してる感があるのでこっち使います

とりあえず実装したものが以下のものです

app.py
from flask import Flask, jsonify, request
from memos.get_memo import get_all_memos as get_all_memos_from_db
from memos.get_memo import get_memo as get_memo_from_db


@app.route("/memos", methods=["GET"])
def get_all_memos():
    all_memo = get_all_memos_from_db()
    return jsonify(all_memo)


@app.route("/memos/<id>", methods=["GET"])
def get_memo(id: str):
    memo = get_memo_from_db()
    if memo is None:
        return {"message": "memo not found"}, 404

    return jsonify(memo)

ここで"/memos/<uuid:id>"についてです.

You can add variable sections to a URL by marking sections with . Your function then receives the as a keyword argument. Optionally, you can use a converter to specify the type of the argument like .

https://flask.palletsprojects.com/en/3.0.x/quickstart/#variable-rules

というわけでURLをいい感じに値を入れられます.その際に型も指定できます.
指定できる型はstring int float path uuidのようです.

メモの区別はuuidなのでuuidを指定しますしたいところですが,JSONで保存してるので型はstrです.

これでメモの取得ができるので,http://127.0.0.1:5000/memosとかに接続して試してみてください.

client側

client側で注意すること2つです.

  • 取得したメモはobjectの配列になっていること
  • メモを作るとserver側からidが返ってくること

server側からidが返ってくるやつは,そこでもう一度すべてのメモを取得するっていうのでも一応できます.

後は気合で何とかしてください

ユーザー認証

さて,ユーザーごとに区別してメモを保存していきましょう

最近使ってるっていう理由で,firebaseで認証します

google認証を使います

ちなみに

Firebase サービスの API キーは非公開ではない
https://firebase.google.com/support/guides/security-checklist?hl=ja#understand_api_keys

らしいですね

ユーザーの区別

Firebase クライアント アプリがカスタム バックエンド サーバーと通信する場合、そのサーバーに現在ログインしているユーザーを特定する必要が生じる場合があります。これを安全に行うために、正常なログイン後、ユーザーの ID トークンを HTTPS を使ってサーバーに送信します。次に、サーバー上で ID トークンの完全性と信頼度を確認し、ID トークンの uid を取得します。サーバーで現在ログインしているユーザーを安全に特定するために、この方法で送信された uid を使用できます。

というわけで,firebase authではIDトークンを検証てログインしている人を特定します

IDトークン検証のためのデコレータ作成

client側でfetchするときにAuthorization headerにIDトークンを載せてserver側に送信し,server側でそのIDトークンの人を検証し特定します.

この辺の処理をスマートにやっている記事がありますので以下を参照してください.こちらはJWT認証なので,その認証している部分をfirebaseの認証に切り替えます
firebase-adminをインストールする必要があります

デコレータやfunctiollsのwrapsに関しては上の記事に加え,以下の記事を参考にしました

そのデコレータを作って適用させてあげると,第1引数にユーザーIDが入るようになります.
最終的に出来上がったものが以下のものです.

verify.py
from functools import wraps

from firebase_admin import auth
from flask import request


class AuthError(Exception):
    pass


def get_token_auth_header():
    auth = request.headers.get("Authorization", None)
    if auth is None:
        raise AuthError("Authorization header is expected")
    auth_parts = auth.split()
    if auth_parts[0].lower() != "bearer":
        raise AuthError("Authorization header must start with Bearer")
    elif len(auth_parts) == 1:
        raise AuthError("Token not found")
    elif len(auth_parts) > 2:
        raise AuthError("Authorization header must be bearer token")

    token = auth_parts[1]
    return token


def verify_token(token):
    try:
        decoded_token = auth.verify_id_token(token)
        return decoded_token
    except Exception:
        raise AuthError("Invalid token")


def require_auth(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        try:
            token = get_token_auth_header()
            decoded_token = verify_token(token)
        except AuthError as e:
            return {"message": str(e)}, 401
        except Exception as e:
            return {"message": f"Internal server error. {str(e)}"}, 500

        uid = decoded_token.get("uid")

        return f(uid, *args, **kwargs)

    return wrapper

firebaseの認証

firebaseで認証するとき,Firebase Admin SDKを初期化する必要があります.
この記事ではアプリをデプロイしないので,Firebase Admin SDKを初期化するにはFirebase サービス アカウントの秘密鍵のパスを指定する必要があります.詳細は以下を見てください

秘密鍵を作成したらGOOGLE_APPLICATION_CREDENTIALSという環境変数にそのパスを指定してあげます
そしてちゃんと現在のシェルで環境変数に値が入っているか確認しましょう.ない場合はexportで値を入れるなり,sourceで読み込むなりしてください

そして,app.pyで初期化してあげれば,IDトークンの検証の下準備が完了です.

app.py
from firebase_admin import initialize_app
from flask import Flask, jsonify, request
from memos.get_memo import get_all_memos as get_all_memos_from_db
from memos.get_memo import get_memo as get_memo_from_db
from memos.save_memo import save_memo
from util.verify import require_auth

app = Flask(__name__)
firebase_app = initialize_app()

# 略

ユーザーごとのメモを取得

先ほど頑張ってuser_idを取得できるようになりました.
なので今度はユーザーごとのメモを取得を取得しましょう.
/users/user_id/memosGETするとメモの一覧が取れるようにします.
/memosにユーザーの検証してGETすれば良くね?

まずはメモ自体にuser_idの情報を入れます.メモを保存するときに入れましょう.
つまり/memosPOSTに認証機能を付けます

app.py
from util.verify import require_auth

app = Flask(__name__)
firebase_app = initialize_app()


@app.route("/memos", methods=["POST"])
@require_auth
def index(user_id: str):
    req: dict = request.get_json()
    memo = req.get("memo")
    id = save_memo(memo, user_id)
    if id:
        return {"id": id}, 200
    else:
        return {"message": "memo have no content"}, 400
save_memo.py
import json
import uuid

from util import constant


def save_memo(memo: list[str], user_id: str) -> bool:
    if not memo or len(memo) == 0:
        return None

    memos = None
    id = str(uuid.uuid4())
    with open(constant.MEMO_FILE, "r") as f:
        try:
            memos = json.load(f)
        except json.decoder.JSONDecodeError:
            memos = []
        memos.append({"memo": memo, "id": id, "user_id": user_id})
    with open(constant.MEMO_FILE, "w") as f:
        json.dump(memos, f, indent=2, ensure_ascii=False)

    return id

あとはメモを取得する関数でuser_idでのフィルタリングをしてルートを作りましょう.
あと忘れてましたが,メモの保存は/users/<user_id>/memosPOSTでやるのでそこも修正します.

get_memo.py
import json

from util import constant


def get_all_memos(user_id: str = None):
    memos = []
    with open(constant.MEMO_FILE, "r") as f:
        try:
            memos = json.load(f)
        except json.decoder.JSONDecodeError:
            memos = []

    if user_id is None:
        return memos

    user_memos = list(filter(lambda m: m["user_id"] == user_id, memos))
    return user_memos


def get_memo(memo_id: str, user_id: str = None):
    memos = get_all_memos(user_id)
    memo = next(filter(lambda m: m["id"] == memo_id, memos), None)
    return memo
app.py
@app.route("/users/<user_id>/memos", methods=["GET"])
@require_auth
def get_user_memos(uid, user_id: str):
    if uid != user_id:
        return {"message": "Unauthorized"}, 401

    memos = get_all_memos_from_db(uid)

    return jsonify(memos)

client側でデータ取得

client側では,サインインしていない状態だとすべてのメモを見れて,サインインしていると自分のだけ見れるようにします.
アプリとしては意味わからんですけど

まあなんかいい感じにサインインさせてuserが切り替わるごとにfetchすればいいと思います

サインインしてない状態

image.png

サインインした状態

image.png

というわけで出来上がったものがこちらです.Chakra UI使いました

image.png

終わりに

この記事では,PythonでFlaskを使ったREST API(風味)を作りました
また,Firebase Authenticationを使いユーザー認証をデコレータを使っていい感じに認証させることができました

Flaskについてがメインっていったけどちょっと薄味になったかもしれないです.
ドタバタで作ったので変なこと言ってたりしたらごめんなさい

それはそれとして,BunとかRyeとか触れたので楽しかったです.

  1. 見つけてくれたapp.pyの場所は試した感じ,作業ディレクトリ直下,srcという名前のディレクトリ内だけでした.src/serverみたいな2階層下だと認識してくれませんでした.どっかにちゃんと書いてたりするんですかね.

1
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
1
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?