LoginSignup
1
3
@TechForGood(Kawamura Yuki)

三週間で駆け抜けたバックエンド個人開発 - docker, mysql, fastapi（jwt認証付き） -

Last updated at Posted at 2024-04-11

1. はじめに

Next.jsによるフロントエンド開発がひと段落した後、サーバーサイドにも手を広げてみたいと考えていました。
サーバーサイドの言語やフレームワークは多様ですが、堅牢且つスピーディーにAPIの実装が可能であるFastAPIに魅力を感じ、学習と試作を進めたので、今回は忘備録も兼ねて経過をまとめていきたいと思います。

この記事では、Docker、FastAPI、MySQLについてそれぞれ前提知識のところから解説を始めているので、結構長くなっています。
また、解説の形はそれぞれの技術の基本や操作方法を紹介するにとどめており、ハンズオン形式でアプリを作ってみようというものではありません。
手っ取り早く具体的なコードだけ知りたい、ハンズオン形式で自分でひとつ作ってみたいという方は、別の記事を読むことをお勧めします。

目次

2. アプリの概要

【制作物】
読書管理（積読対策）アプリ

【技術スタック】

APIサーバー
fastapi
データベース
MySQL
仮想環境構築
Docker

※元々フロントエンドにはReact（Next.js）を採用するつもりでしたが、今回はサーバーサイドの勉強に専念したいと思い、サーバーの実装を中心に取り組みました。

【ディレクトリー構成】

C:.
│  .env
│  .gitignore
│  docker-compose.yml
│  Dockerfile
│  requirements.txt
│
└─app
    │  db.py
    │  main.py
    │  migrate_db.py
    │  __init__.py
    │
    ├─cruds
    │  │  archived_book.py
    │  │  auth.py
    │  │  book.py
    │  └─ __init__.py
    │
    ├─models
    │  │  archived_book.py
    │  │  book.py
    │  │  user.py
    │  └─ __init__.py
    │
    ├─routers
    │  │  archive_bm.py
    │  │  auth.py
    │  │  book_manager.py
    │  └─ __init__.py
    │
    ├─schemes
    │  │  book.py
    │  │  res_msg.py
    │  │  token.py
    │  │  user.py
    │  └─ __init__.py
    │
    └─utils
       │  error.py
       │  jwt.py
       └─ pwd.py

全体のプロダクトコードはGithub上に公開しています。

サーバー側の作成手順は概ね以下の二つの記事を参考にさせていただきました。

特にFastAPI入門ではTodoアプリの制作をハンズオン形式で解説されてあります。
私のアプリの場合、コーディングは参考元と多少変わっていますが構造としては似たようなものなので、FastAPIに関わる部分では異なるコードのところを中心に解説していこうと思います。

3. Docker

以降の解説では、Dockerがインストールされている前提で話を進めます。

Node.jsのパッケージ管理ツールであるnpmは、package.jsonファイルが存在するディレクトリーをルートディレクトリ―として、package.jsonで依存関係が宣言されてあるパッケージをローカルにインストールします。
ローカルのパッケージインストール先はnode_modulesになるわけですが、このようにnpmはパッケージ/ライブラリーをグローバルにインストールするわけではなく、あくまでもそのプロジェクトごとに、package.jsonが配置されてあるディレクトリー配下にインストールします↗。

一方でpythonの一般的なパッケージ管理ツールであるpipは、デフォルトではパッケージをグローバルにインストールします。
そのため、何も考えずrequirements.txtにしたがってインストールしていると、別のプロジェクトには不必要なパッケージがインストールされてグローバルなPythonシステムが汚染されてしまったり、パッケージのバージョンの違いにより依存関係が解決できなくなったり、anacondaと衝突して機能不全に陥ったりします↗。

Node.jsにおけるnode_modulesのように、関係するパッケージのインストールをプロジェクトを管理するディレクトリーの内部で完結させたいとき、pythonの仮想環境構築コマンド python -m venv [任意の仮想環境名称] を使うことが多いと思います。
しかし、私自身そうなのですが、仮想環境をactivateしてからpip installする手順をうっかり飛ばしてしまい、結局グローバルインストールしてしまうことがありました。

これは厄介な問題です。
npmと違い、pipはデフォルトがグローバルインストールであるために、仮想環境外でpip installしてしまうヒューマンエラーは普通に起こり得ます。
このような背景で、特にPythonにおいてはローカルなパッケージ管理をしたいというとき仮想環境の構築方法としてDockerも候補に挙がることがあります。
FastAPI入門においても、環境差分の排除、環境の閉じ込めという点でDockerの利点は語られています。これに加えて上記のように、pythonの仮想環境にまつわるヒューマンエラーをなくすという点でも、Dockerの利用はお勧めできます。

3-1. Dockerの概要

これ以降のDockerに関する解説は、基本的に鈴木亮様の開発系エンジニアのためのDocker絵とき入門を参考にしています。

開発環境を仮想化する技術は、次のように三つに分かれると言います。

  • ホスト型仮想化
  • ハイパーバイザー型仮想化
  • コンテナ型仮想化

このうち、Dockerはコンテナ型仮想化にあてはまります。
コンテナ型仮想化とは、コンテナという単位でアプリケーションを管理する手法のことで、ゲストOSをもたないものの、ホストマシンの機能を利用することでコンテナーの土台としてゲストOSが機能しているように見せかけることができます。この仮想的なOSの上に各種プログラムや実行環境をのせることで、依存関係や実行環境が閉じられた状態（コンテナー）をつくることができます。

3-2. Docker コマンド

dockerのコマンドは、dockerに続くコマンドと、docker containerのように操作対象を表すサブコマンドに続くコマンドがあります。

Dockerのコンテナーやイメージを扱う主要なコマンドはサブコマンドにまとめられており、主なものとして次の四つがあげられます。

  1. image
  2. container
  3. volume
  4. network

これらのサブコマンドに対し、共通して四つのコマンドが存在します。

  • ls： 一覧表示
  • inspect：詳細表示
  • rm：削除
  • prume：未使用のものをすべて削除

このうち、network以外を軽く解説しようと思います。


Dockerは、配布されているイメージからコンテナーを作り、そのコンテナーの中でプログラミングを進めていきます。

Dockerにおけるイメージはコンテナーの基本設計のようなもので、複数のレイヤとよばれるtarアーカイブファイルの積み上げで成り立っています。
同じイメージから作られるコンテナーは同一である一方で、コンテナーに対しておこなわれた操作はイメージには反映されません。
オブジェクト指向プログラミングにおける、クラスとインスタンスの関係に近いといえるかもしれません。

Ⅰdocker image
docker imageのサブコマンドの中で、よく使われそうなものは次の二つです。

  • pull：Docker hubからイメージを取得
  • build：Dockerfileからイメージを作成

・docker image pull [イメージ:tag]
これは、Docker Hub上などから、MySQLやRubyなど公式のイメージを探し、タグを指定して取得してきます。
タグはイメージのバージョンやタイプを指定するものだと思ってもらって構いません。
タグを指定しなければ最新バージョンがインストールされることになります。

pullが「引っ張る」という意味なので、インターネット上に公開されてあるイメージを引っ張ってきて手に入れるというイメージにぴったりのコマンド名ですね。

・docker image build [DockerfileのあるパスまたはURLなど]
Doker Hub上に公開されてあるデフォルトのイメージをカスタマイズしたい時に使えます。

コンテナーのカスタマイズをしようとすると通常は、image pullでイメージを取得した後、docker container に続けて詳細なコマンド、オプションの設定が必要です。
しかしコンテナーに沢山の設定を加えたとき、それを他の人と共有したり、コンテナーを再び立ち上げたりするのは大変です。沢山のコマンドやオプションをすべて正確に記述しないといけないからです。
コンテナー対して追加した設定をイメージとして固定して簡単に持ち運べるようにしたいとき、Dockerfileが有用です1。

image buildコマンドは、Dockerfile内に記述された命令文を実行することによってカスタマイズされたイメージをつくることができます。

例えば、今回のプロダクトにおけるDockerfileは以下の通りです。
（FastAPI公式で示されているDockerfileとほぼ一緒です）

FROM python
 
WORKDIR /code

COPY ./requirements.txt /code/requirements.txt
 
RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt
 
COPY ./app /code/app
 
CMD ["uvicorn", "app.main:app","--reload", "--host", "0.0.0.0", "--port", "8000"]

其々の命令文を解説します。

FROM
ベースイメージの指定。
WORKDIR
Dockerfile内で実行されるRUN、CMD、ENTRYPOINT、COPY、ADD命令の作業ディレクトリを設定します。つまり、コンテナ内でこれらの命令が実行される場所を指定します。デフォルトでは / (カレントディレクトリー)で、特に指定がなくても正常に機能することがありますが、作業ディレクトリーは明示することが推奨されています。
COPY
ホストマシンのファイルをイメージにコピーする。
RUN
コマンドを実行し、レイヤー（≒イメージの一部）として確定する。
CMD
コンテナ起動時に実行するコマンドのデフォルト部分を指定する。

注意するべきポイントは二点あります。
まずは五行目のCOPY命令文についてです。
Dockerはビルドの際、内部キャッシュを利用します。既にビルドされたことがあるならば、ビルド時の情報はイメージにレイヤーとして組み込まれており、特に変更がない限り再利用されます。
requirements.txtを作業ディレクトリー(/code)にコピーして、RUN pip install --no-cache-dir --upgrade -r /code/requirements.txtとしているのは、キャッシュを利用するためです。

pip install -r [依存関係を記述したスクリプト]を実行する際、イメージにコピーしたrequirements.txtの代わりに、ホストマシン側のrequirements.txtを指定すると、ビルド時にわざわざ一からパッケージを一括インストールしていることがわかります。コンテナーの立ち上げに時間がかかるわけです。
必要となるパッケージはあまり変更されないはずですから、キャッシュを利用するのが安牌でしょう。

次に、最終行のCMD命令文についてです。
これはコンテナー起動時のデフォルトのコマンドを指定するもので、今回は"uvicorn"でサーバーを起動するようにしています。
--reloadで、ホットリロードの機能を実装し、--host, --portでホストマシンからアクセスできるサーバーのURLを指定しています。

Ⅱdocker container
docker container コマンドのうち、よく使われそうなのは以下の通りだと思います。

  • run
  • stop
  • start
  • rm
  • logs
  • exec

・docker container run [コンテナー元のイメージ]
これは、イメージをもとにコンテナーを立ち上げるコマンドです。
このコマンドは、create、start、attachの三つのコマンドを一気に実行するものであり、具体的にはコンテナー作成、起動、ターミナルのコンテナーへの紐づけの処理を行っています。

また、runコマンドを実行する時、幾つかのオプションをつけることができます。よく使うオプションは以下の通りです。

コマンド 意味
--name コンテナーの名づけ
--rm 終了したコンテナーの自動削除
--interactive コンテナーの入力に接続
--tty 疑似ターミナルの割り当て
--publish コンテナーのポートをホストマシンに公開
--env 環境変数の設定
--detach コンテナーの出入力とターミナルを切り離す

表の補足を少しだけします。
--interactive と --ttyはあまり見たことがないかもしれませんが、-itや-tiの形であれば見たことがあるかもしれません。
二つを指定することによって、runコマンドを実行したプロンプトにおいて対話形式でプログラムを打ち込むことができます。Pythonであれば対話シェルが動きます。この二つのオプションについてより深く理解しようとするとかなり長くなるので、他の方の解説をご参照ください↗。

--detachは、バックグラウンドでコンテナーを実行する際に使います。
先述した通り、docker container run コマンドは、create、start、attach を一気に実行します。attachコマンドの実行処理が含まれるために、runコマンドを実行するとターミナルがコンテナーの出入力と結びつき、それ以上そのターミナルを操作することができなくなります。
あえてそうする必要もありませんが、container create、container start の二つを続けて実行することでも、--detachオプションと同じことができます。

・docker container logs [対象コンテナー]
--detachオプションをつけてrunコマンドを実行すると、コンテナー内のアプリの実行結果が見えなくなります。
Docker Desktopでコンテナーのログをみる方法もありますが、もしコマンドプロンプト上でコンテナーのログ出力をみたいようであれば、logsコマンドを実行すればそれが可能です。logsコマンドのオプションとしては、--followオプションがあり、これはログを永久出力させることができます。

・docker container exec [対象コンテナー] [コンテナー内プログラムに対して実行してほしいコマンド]
これは起動中のコンテナーに対して、コンテナー内で新しいコマンドを実行させることができます。

Ⅲ docker volume
dockerにおいて、コンテナーのデータをコンテナーの外とつなげたいときがあります。
それは主に2パターンあって、ひとつめがコンテナー内のデータを永続化させたいとき、ふたつめがコンテナーに対してホストマシンのディレクトリーをマウントさせたいときです。

コンテナー内のデータを永続化させたいときは、docker volume create でボリュームを作る際に、--mount type=volume,source=(マウント元),destination=(マウント先) の形でmountオプションを指定することで可能です。例えば、MySQLのコンテナーを削除した後もデータベースのデータを保持したままにしたいときや、他のコンテナーからもそのデータにアクセスしたいときなどは、mountオプションの指定を--mount type=voluem,source=mysql-data,destination=/var/lib/mysql とします。

コンテナーに対してホストマシンのディレクトリーを結び付けたいときは、moutオプションのtype指定を「bind」にするだけで大丈夫です。これにより、ホストマシン上の作業ディレクトリーの変更がコンテナーに反映されることになります。

これで、主要なコマンドの解説が終わりました。

3-3. docker compose

Dockerのコンテナーは、Docker HubからpullするにしろDockerfileをbuildするにしろ、ひとつのプログラムしか抱えられません。
このため、例えばAPIサーバーとMySQLデータベースを備えたバックエンドアプリを開発しようとすると、其々のコンテナーを作成・起動したうえで、docker networkでコンテナー間の通信を確立しなければなりません。これは非常に面倒です。
そのような時に役に立つのが、docker composeというシステムであり、コマンド一つで複数のコンテナーを立ち上げ、相互の通信を確立することができます。

docker composeで用いられるプロパティはこれまでに紹介して来たコマンドやオプションが使われているため、その意味を理解していればdocker composeを使う時にはあまり困らないと思います。

設定ファイルを見てみましょう。

docker-compose.yml
version: "3"

services:
  app:
    container_name: "app"
    environment:
      MYSQL_DATABASE: "fast"
      MYSQL_USER: "practice"
      MYSQL_PASSWORD: "practice-db"
      MYSQL_ROOT_PASSWORD: "practice-db"
    volumes:
      - type: bind
        source: ./app
        target: /code/app
    build: .
    ports:
      - "8000:8000"

  db:
    image: mysql
    environment:
      MYSQL_DATABASE: "fast"
      MYSQL_USER: "practice"
      MYSQL_PASSWORD: "practice-db"
      MYSQL_ROOT_PASSWORD: "practice-db"
      TZ: "Asia/Tokyo"
    volumes:
      - type: volume
        source: mysql_data
        target: /var/lib/mysql
    ports:
      - "3306:3306"

volumes:
  mysql_data:

前節でDockerのコマンドを解説していったのは、上記の記述を理解してもらうためです。Dockerのコマンドとファイルのプロパティは対応しており、なんとなく雰囲気はつかめると思います。

appコンテナーでは、Dockerfileがdocker-compose.ymlファイルと同階層にあるのでカレントディレクトリー指定でビルド(build: .)しています。
一方でdbコンテナーではDockerfileは用いず、mysqlイメージをpullすることによってコンテナーを作成しています。

servicesプロパティと同じトップレベルにvolumesプロパティがあると、なんとなくトップレベルのvolumesが浮いて見えるかもしれません。
これは、データの永続化のために行った、docker volumes createコマンドに相当します。前節で解説した通り、コンテナー内のデータを永続化するためには、Docker Engineが管理する領域にディレクトリーを作成し、そこにたいしてマウントしなければなりません。マウントの実行はdbプロパティ内のvolumesプロパティで指定していますが、そのままではそもそもデータの保存先が作られていないため、トップレベルのvolumesプロパティでデータの保存領域を確保してあげる必要があった訳です。

docker compose のコマンドは以下の通りです。

  • docker compose up
  • docker compose start
  • docker compose exec
  • docker compose stop
  • docker compose down

・docker compose up [オプション] [サービス名]
upコマンドを実行する際は、docker composeを記述したymlファイルを指定することもありますが、作業ディレクトリー内にcompose.ymlやdocker-compose.ymlなどの名前でymlファイルがあれば、わざわざ明示しなくてもdocker compose upとコマンドを叩くだけでサービスを立ち上げることができます。
upコマンドでよく使うオプションは以下の二つだと思います。

オプション 意味
--detach バックグラウンド実行
--build コンテナー起動前にイメージをビルドする

・docker compose down [オプション] [サービス名]
compose.ymlファイルがあるディレクトリーで実行すれば、特にサービス名を明示しなくていいのはdocker compose upコマンドと同じです。
downコマンドは、コンテナーを削除するコマンドですが、ボリュームやイメージにはノータッチです。
Docker Desktopからでも各イメージなどは削除できますが、コマンド上でそれらを削除できる便利なオプションがあります。
もしイメージやボリュームまで削除したいのであれば、docker compose down --rmi all --volumesと指定してあげれば大丈夫です。


一つ注意したいのは、上記のdocker-compose.ymlファイルの例では環境変数が直書きされているということです。これはあまり好ましくありません。
そこで、環境変数を.envファイルにまとめ、それをdocker-compose.ymlファイルに取り込むようにします。
例えば、以下の例です。

docker-compose.yml
version: "3"

services:
  app:
    container_name: "app"
    environment:
      MYSQL_DATABASE: ${MYSQL_DATABASE}
      MYSQL_USER: ${MYSQL_USER}
      MYSQL_PASSWORD: ${MYSQL_PASSWORD}
      MYSQL_ROOT_PASSWORD: ${MYSQL_ROOT_PASSWORD}
    volumes:
      - ./app:/code/app
    build: .
    ports:
      - "8000:8000"

  db:
    image: mysql
    environment:
      MYSQL_DATABASE: ${MYSQL_DATABASE}
      MYSQL_USER: ${MYSQL_USER}
      MYSQL_PASSWORD: ${MYSQL_PASSWORD}
      MYSQL_ROOT_PASSWORD: ${MYSQL_ROOT_PASSWORD}
      TZ: "Asia/Tokyo"
    volumes:
      - mysql_data:/var/lib/mysql
    ports:
      - "3306:3306"

volumes:
  mysql_data:

pythonの側（appコンテナー）でも環境変数を設定しているのは、Pythonファイルの中で、osを用いて環境変数を取り込みたいからです。
上記の例ではdocker-compose.ymlファイルと同じ階層に.envファイルを設定しました。
.envファイルは同階層に設定するのが一番わかりやすいと思いますが、もし環境変数の数が多かったり、envファイルがサブディレクトリーの中にある場合などは、env_fileプロパティを用いることで、相対ルートで指定された環境変数ファイルを取り込むこともできます。


これで、FastAPIをapiサーバーとし、MySQLをデータベースとするバックエンドアプリの環境構築が終了しました。
次は、FastAPIの解説に入ります。

4. FastAPI

FastAPIとは、Pythonの軽量なWebアプリケーションフレームワークのことです。
Node.jsやGoに匹敵するほどの高速なアプリケーションの開発が可能で、SwaggerUIと連結しているので、開発を進めれば自動的にAPIドキュメントを作ってくれます。
FastAPIがPythonのフレームワークであるため、Pythonに豊富な機械学習フレームワークとの相性が良いという点も魅力の一つです↗。

第四章では、FastAPIについて解説していきます。

4-1. APIとは

この節の解説は主に、WebAPI The Good Partsという書籍を参考にさせていただきます。

APIとは、「Application Programming Interface」の略で、アプリケーション同士が情報をやり取りする仲介役のようなイメージになると思います。
その中でもWebAPIとは、アプリケーション同士のやり取りのプロトコルとしてHTTPを利用するようなAPIのことです。

APIは特定のサービスのクライアント － サーバー間でやり取りする専用に開発されたものもあれば、広く一般に公開されるAPIもあります。
一般に公開されたWebAPIの成功例としては、Amazonや旧Twitterがあげられます。たとえば、AmazonがAPIを公開したことでさまざまな外部サービスが立ち上がり、アフィリエイトで商品を購入するスタイルが生まれました。

FastAPIの解説に入る前に、APIの基本的な知識として、以下の三つについて軽く解説を加え異様と思います。

  1. エンドポイントの基本的な設計
  2. HTTPメソッドの種類とその意味
  3. パスかクエリーか

　Ⅰ：エンドポイントの基本的な設計
APIのエンドポイントはURIと一致するので、よいエンドポイントとはほぼ、そのままよいURI設計を意味します。
よいURI設計のための要素は以下の通りです。

  • 短く理解しやすいURI
  • ルールが統一されたURI
  • 改造しやすいURI

URIが短くて理解しやすいものであるべきなのは、特に論を俟たないでしょう。
ルールが統一されたURIとは、URIの文字列が大文字・小文字のどちらかに統一され、単語をつなげる際はキャメルケースやハイフン、アンダーバーなどのどれかに統一され、詳細情報を取得する際のURIはhttps://api/book/5かhttps://api/book/?id=5のようにパスかクエリー検索かで統一されている必要があるということです。
改造しやすいURIとは、URIを修正して別のURIにするのが容易であることを示しており、これは例えば、https://api/book/5のように、URI末尾の数字を変えれば他の本の情報が手に入ると容易に想像できるような状態を指します。

　Ⅱ：HTTPメソッドの種類とその意味
よく使われるHTTPメソッドは以下のようなものがあります。

メソッド 意味
GET データの取得
POST データの登録
PATCH 既存データの部分更新
PUT 既存データの更新（上書き）
DELETE データの削除

名称とその意味は直感的なので、あまり解説は要らないかもしれません。
一つ注意したいのが、PUTとPATCHの違いです。
PATCHは2010年に発行されたRFC5789（HTTPの定義）において定義された比較的新しいメソッドです。
PUTは既存のデータを上書きしたり、データがなければ新しく登録したりするのに対して、PATCHは既存データの一部分を変更するという意味を持ちます。
上書きをするというとき、データの量が多ければ多い程更新処理に時間がかかりますし、効率的ではありません。そのため、処理速度に優れる「部分的な更新」を意味するPATCHメソッドを使う方が多いと思います。

　Ⅲ：パスかクエリーか？
基本的には、URIが一意に定まる場面においては、URIの記述のしやすさからもパスで指定する方がよいと思います。
しかし、複数の条件を組み合わせたいときや、デフォルトの挙動を設定したいときなどは、クエリパラメータが有用だと思います。
前者について言えば、例えば、https://api/book/5/whole_page/50/...などとするよりは、https://api/book/5/?whole_page=500&...などとする方がわかりやすいでしょう。
後者について言えば、バージョン指定のときがわかりやすいと思います。デフォルトでは最新バージョンを返し、クエリーパラメータで特定のバージョンを返したい、というような振る舞いをさせるときはパスではなくてクエリー検索の方が有用だと思います。例えば、クエリパラメータを省略してhttps://api/book/5/というURIにアクセスするときは最新版の情報が、https://api/book/5/?edition=4とするときは第四版の情報が返される、というような場合です。

WebAPIに関する話はこれくらいにして、FastAPIの解説に入っていきたいと思います。

FastAPIを動かすためには最低限、fastapiライブラリーと、サーバーを動かすためのuvicornライブラリーの二つが必要になります。
requirments.txtで一括インストールしても、Dockerfileのなかで個別にpip installしても大丈夫です。

4-2. ディレクトリー構成

私が作ったアプリの構成は以下の通りです。

C:.
│  .env
│  .gitignore
│  docker-compose.yml
│  Dockerfile
│  requirements.txt
│
└─app
    │  db.py
    │  main.py
    │  migrate_db.py
    │  __init__.py
    │
    ├─cruds
    │  │  archived_book.py
    │  │  auth.py
    │  │  book.py
    │  └─ __init__.py
    │
    ├─models
    │  │  archived_book.py
    │  │  book.py
    │  │  user.py
    │  └─ __init__.py
    │
    ├─routers
    │  │  archive_bm.py
    │  │  auth.py
    │  │  book_manager.py
    │  └─ __init__.py
    │
    ├─schemes
    │  │  book.py
    │  │  res_msg.py
    │  │  token.py
    │  │  user.py
    │  └─ __init__.py
    │
    └─utils
       │  error.py
       │  jwt.py
       └─ pwd.py

ルートディレクトリー内のDockerfileやdocker-compose.yml、requirements.txtは前章で解説した通りです。この節では、各サブディレクトリーの意味を話していこうと思います。

crudsディレクトリーは、データベースと連携するファイルを配置しています。
modelsは、データベースのデータ構造を設計しています。
routersは、ルーターの定義を行っています。本当に小さなアプリケーションであれば、main.pyですべてのルーティングを設定することもできます。しかしルートを機能ごとにファイルに分割したのは、アプリが大きかったり、今は小さくとも今後スケールしたりする可能性があるためです。
schemesは、リクエストやレスポンスに対しての型定義を行っています。pydanticの機能を用いるには、Pythonの普通のクラスではなく、BaseModelを継承した型定義用のクラスを用いる必要があります。

4-3. ルーティング

booksルートに絞って話します。
まず、bookパスに対応させるようなルーティングを設定します。

routers/book_manager.py
from fastapi import APIRouter, status
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List, Union

router = APIRouter()

@router.get("/api/books")
async def books():
    pass
    
@router.post("/api/books",status_code=201)
async def create_book():
    pass 

@router.get("/api/books/{book_id}")
async def book(book_id:int):
    pass

# ... patch delete　メソッドの定義

特定のパスに対して、メソッドと処理内容を記述する関数のことをパスオペレーションといい、これはPythonの「デコレーター」という機能を使います。

引数として他の関数を受け取って内部の処理に組み込む関数（高階関数）を実装するとき、普通にやれば、取り込みたい関数ごとにその関数を引数にして実行する必要があります。
デコレーターとは、高階関数を利用した手順を簡略化するシンタックスシュガーです。一度高階関数を定義しておけば、引数として渡したい関数の上に@methodと記述するだけで自動的に直下の関数が引数として渡され、デコレートされた関数を実行するだけで高階関数の機能が実装されます。
デコレーターの挙動に関してはサプーさんが詳しい解説をしているので、ぜひ参考にしてください。

このほかに二つポイントがあります。
・status_code=201
デフォルトでは成功した場合のステータスコードは200ですが、関数の引数としてstatus_codeを指定することで変更することができます。
・@router.get("/api/books/{book_id}")
動的ルーティングは@router.get("/api/books/{book_id}")のようにパスに埋め込んだ後、async def book(book_id:int):のように関数の引数として指定するだけで実装できます。クエリパラメータはパスに何も記述せずに、関数の引数として設定するだけで実装可能です。

main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.routers import auth, book_manager, archive_bm

app = FastAPI()

origins = [
    "http://localhost:3000",
]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(auth.router)
app.include_router(book_manager.router)
app.include_router(archive_bm.router)

FastAPIをインスタンス化したappインスタンスのinclude_routerメソッドに、引数としてルーターを渡すことで、他ファイルで定義したルーティング機能を実装することができます。

add_middlewareというメソッドはその名の通りミドルウェアを実装しています。何かアプリ全体で共通の設定を行いたい場合は、このメソッドの引数として様々設定することも可能です。

4-4. スキーマ

bookスキーマに絞って解説します。

schemes/book.py
from typing import Union
from datetime import datetime
from pydantic import BaseModel, Field

class BaseBook(BaseModel):
    whole_pages: int = Field(ge=1)
    read_pages: Union[int,None] = Field(default=0)
    created_at: datetime = Field(default=datetime.now())
    deadline: Union[datetime,None] = Field(default=None, description="The date when you want to read the book by.")
    title: str = Field(..., example="JUSTICE -what the right thing to do?-")
    comment: Union[str, None] = None
    to_archive: bool = Field(default=False, description="If you finish reading the book, check the box. This book will move to archive database.")

class EnrollBook(BaseBook):
    pass

class ModifyBook(BaseModel):
    whole_pages: Optional[int]
    read_pages: Optional[int]
    created_at: Optional[datetime]
    deadline: Optional[datetime]
    title: Optional[str]
    comment: Optional[str]
    to_archive: Optioanl[bool]

class CreateBook(BaseBook):
    user_id: int = Field(ge=1)

class DataBaseBook(CreateBook):
    id: int = Field(ge=1)

class ResponseBook(DataBaseBook, user.ConcatUserFromBook):
    pass

pydanticにおいて、型定義の大本になるクラスがBaseModelであり、Fieldは、デフォルトの値やその制限(数値の制限や正規表現)、そしてプロパティを説明する補足的な情報(titleやexample)を提供することができます。

度々登場するFieldの設定、「ge=1」とは、1以上という意味です。数値の制約は幾つかあるのでここで少し紹介します。

数値制約 意味
ge >=
gt >
le <=
lt <

プロパティがオプショナルであることを示す方法は二つあり、Optional[type]か、Union[type,None]です。前者はそのプロパティが必須ではなくなってNoneも受け取るようになり、後者はそのプロパティが指定された型かNoneかの選択制になります。ほぼ同じ挙動です。

詳説　UnionとOptional

pydantic v1においては、オプショナルなプロパティの指定として、Union[type,None]と、Optional[None]の二通りが許容されています。一方pydantic v2に移行すると、python 3.10以降の記法type | None = Noneのような記述が推奨されています。

とはいえどちらにしても、プロパティの値としてNoneが許容されているため「そのプロパティが渡されなくてもいいけれど、渡されたときはNoneを許容しないでほしい」という場合は、上記の方法では不十分です。
この場合クラス内で、validatorデコレーターを用いて、プロパティが渡された場合はNoneではないという検証が必要になります。
このことはGithubのissueにも挙がっているので、ぜひ参考に指定ください。

How to have an “optional” field but if present required to conform to non None value?

このschemeをルーターに組み込むことで、リクエストボディーのデータ型を検証したり、型を明示してレスポンスを返却することができます。
例えば、以下の通りです。

routers/book_manager.py
from fastapi import APIRouter, status
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List, Union

from app.schemes import book as book_scheme

router = APIRouter()

@router.get("/api/books",response_model=List[book_scheme.ResponseBook])
async def books():
    book = db.info.books()
    # databaseからデータを取得してくる何某かの処理
    return book
    
@router.post("/api/books",response_model=book_scheme.ResponseBook,status_code=201)
async def create_book(book_body:book_scheme.EnrollBook):
    book = db.create.book(book_body)
    # databaseにデータを登録する何某かの処理。
    # 返り値として、idまで含めたbookのデータを取得。
    return book

@router.get("/api/books/{book_id}",response_model=book_scheme.ResponseBook)
async def book(book_id:int):
    book = db.info.book()
    return book

# ... patch delete　メソッドの定義

パスオペレーションの引数として、response_modelを設定します。これを設定しておくことによって、サーバーが返すレスポンスのデータ型が正しいかどうかをチャックし、また、SwaggerUIにおいてResponseのbodyが例示されます。
create_book関数において、book_bodyをEnrollBook型の引数として設定しています。引数の型として、intやstrといった単体のデータ型ではなくスキーマで定義したデータ型を設定する時、それはクエリーパラメータではなくリクエストボディの型と見做され、リクエストボディに対して型の検証が行われます。

リクエストボディのプロパティが少なく単数値を用いたい場合や、どうしても型定義にスキーマを用いたくない場合があるかもしれません。
そんなときは、FastAPIにクエリーパラメータと見分けてもらうために、その引数の型としてname:str = Body()のように、Bodyの依存性注入を行なえば大丈夫です。

5. SQL

SQLの基本概念やSQL文の記法などについては、すっきりわかるSQL入門を参考にしています。

SQLとは、データベースを操作する専用の言語のことです。
データベースそれ自体はただのファイルですので、送られてきたSQL文に基づいてデータベースを操作するためのシステムがないと動きません。
データベースの中のデータを操作するものがデータベース管理システム(DBMS/Database Management System)とよばれるもので、MySQLこはこれの一つです。

DBMSは実行環境を用意してデータベースを操作するコマンドを立ち上げれば、コマンドライン上でSQL文を打ち込むことができますが、実際の開発環境ではアプリとデータベースの接続を確立したのち、何かしらのライブラリーを用いてデータベースへの命令を送ることの方が多いと思います。
命令の形式は、生のSQL文を送ることもありますし、オブジェクト指向プログラミング言語を用いているならば、MySQL（等のリレーショナルデータベース）とプログラミング言語のデータ構造上の差を処理して互換性を高めるORM（object relational mapping）を用いることもあると思います。

5-1. SQLの基本

ここでは、SQLの基本的な記法を解説します。

SQLの体系は以下の通りです。

命令文 各命令に固有の句 検索の条件 検索結果の加工
SELECT [列名] FROM WHERE 修飾句
UPDATE [テーブル名] SET WHERE -
DELETE FROM WHERE -
INSERT INTO [テーブル名] VALUES - -

SQLの命令は、SELECT, UPDATE, DELETE, INSERT の四つしかありません。
私は、データベースはもっと複雑なものと思っていたので、根幹となるデータ処理の方法が四通りしかないシンプルなものだと知って驚きました。
よくよく考えてみればHTTPメソッドも、よく使うものは同じようにGET, POST, PUT/PATCH, DELETEしかないので、大別すればデータの処理の方法なんてそれくらいしかないのでしょう。

上記の命令文の其々について、解説を加えていきます。

　Ⅰ：SELECT文
SELECT文の基本的な形は以下の通りです。

SELECT id, email, username
FROM user
WHERE id = 1;

SELECT文で取り出したいデータの列名を指定し、FROMでどのテーブルからデータを取り出すかを定め、WHEREで取り出したいデーの条件を示します。基本的な形はこの通りで、この下に修飾句が尽きます。

　Ⅱ：UPDATE文
UPDATE文の基本的な形は以下の通りです。

UPDATE user
SET username = "sql-learner"
WHERE id=1;

UPDATEで更新したいuser名を指定し、SETで変更したいデータの列名を指定し、WHEREでどの行のデータを変更するかを指定します。
WHERE句による条件指定がない場合、テーブルのusernameという列のデータ全てが新しいユーザーネームに置き換わってしまうので、必ずWHERE句は付けなければなりません。

　Ⅲ：DELETE文
DELETE文の基本的な形は以下の通りです。

DELETE FROM user
WHERE id=1;

DELETE FROMで削除したいデータのあるテーブルを指定し、WHEREで削除したい行を指定します。
WHEREで条件を絞っていないと、テーブルの中の全てのデータが削除されてしまいます。テーブル自体は削除されませんが中身は空っぽになるので、まかり間違ってもDELETE FROM [テーブル名] だけのDELETE文を実行しないようにしましょう。

　Ⅳ：INSERT文
INSERT文の基本的な形は以下の通りです。

INSERT INTO user
VALUES ("example@gmail.com","sql-learner","@fast5-s6ql-learner");

INSERT INTOで、新しいデータを登録したいテーブルを選択し、VALUESで登録したいデータの中身を列挙します。
上記の例はテーブルの列名を省略した形ですが、この状態で正しくデータを登録するためには、登録データの並びをテーブルの列の順序と同じにしないといけません。
テール部の列の並び順に関わらず値を任意の順で登録したかったり、特定のデータのみを登録したい場合は、テーブル名に続けて列指定もすることができます。
たとえば、以下の例です。

INSERT INTO user (password, email)
VALUES ("@fast5-s6ql-learner","example@gmail.com");

【詳説】
基本的な構文は以上の通りですが、これだけでは表現の幅が狭すぎて、使い物になりません。より実践的なテーブル操作ができるように、他にも様々な記法を紹介します。

・WHERE句
SELECT, UPDATE, DELETE文で用いられるWHERE句は、真偽値が判定できるような形にする必要があります。
具体的な比較演算子としては以下のようなものがあります。

比較演算子 意味
= 両辺が等しい
<= 右辺が左辺以上
>= 左辺が右辺以上
<> 両辺は不等
not 否定
is Null 左辺はNullである
IN 右辺が左辺の集合に含まれる
ANY 左辺のいずれか一つが条件を満たす
ALL 左辺の全てが条件を満たす

また、一つの条件式では、取り出したいデータの条件を定義しきれないことがあります。そのようなときは論理演算子を用いて条件式を結合でき、代表的な論理演算子として、AND演算子（論理積）やOR演算子（論理和）を使うことができます。

このような条件式はWHERE句で用いられるため、INSERT文以外の全てに対して使うことができます。

・検索結果の加工
検索結果を加工する処理は、SELECT命令文のみでしか扱うことはできません。
例えば、以下のようなものがあげられます。

修飾句 加工内容
DISTINCT 重複行の排除
ORDER BY 並び替え
GROUP BY 種類ごとに集計
LIMIT 行数の限定
UNION 結果を集合演算

DISTINCTとUNION句について解説を加えます。

DISTINCTとは、指定した列において重複を排除するものです。Pythonでいえば、例えばnations=["Japan","South Korea","the USA","Japan","Japan","the USA","South Korea"]となっている配列に対して、set(nations)とすると、("Japan","the USA","South Korea")になるのと同じようなイメージだと思います。具体的な記法としては以下の通りです。

SELECT DISTINCT title 
FROM book;

上のSQL文を実行することにより、bookテーブルの中から、本の種類のみを重複なく取り出すことができます。

UNIONとは、二つ以上のテーブルの検索結果を結合するものです。具体的な記法は以下の通りです。

SELECT DISTINCT title 
FROM book
UNION
SELECT DISTINCT title 
FROM archived_book;

これは、bookテーブルと、archived_bookテーブルの二つからそれぞれ本のタイトルを取得してきて結果を返しています。
デフォルトでは重複行を排除していますが、UNION ALLとすれば、重複を許して結果を返却します。
なお、修飾句は結合したSELECT文の後につなげる点に注意です。

・サブクエリー
サブクエリーとは、他のSQL文の一部分として登場するSELECT文のことを示します。丸括弧の中に記述します。
具体的な記法としては、以下の通りです。

SELECT *
FROM book
WHERE "Introduction to SQL" in (SELECT DISTINCT title
                                FROM book)
UNION
SELECT *
FROM archived_book
WHERE "Introduction to SQL" in (SELECT DISTINCT title
                                FROM archived_book)
LIMIT 1;

これは、「Introduction to SQL」という本がテーブルのなかに既に存在するかを判定しています。
サブクエリーは丸括弧内の(SELECT DISTINCT title FROM book/archived_book) の部分で使われており、これは例えば、("Introduction to Python","Introduction to Docker","Introduction to JavaScript","Introduction to Next.js","Introduction to SQL") といった、具体的な検索結果になります。この例でいえば、既に値が存在していることになりますね。

ここではSQL文を紹介するためにこのような長い記法を用いていますが、テーブルにデータが存在するかどうかだけを調べたいのであれば、「EXISTS」を利用する方法もあります。

・テーブルの結合
テーブルは、一つで完結することもありますが、複数のテーブルと関連してデータを保存することもあります。
JOINを用いることで、テーブルから目的のデータを取り出す際に、他テーブルから関連したデータを取り出して連結してから、全体のデータを返却させることができます。

SELECT book.*, user.username, user.email
FROM book
JOIN user
ON book.user_id = user.id
WHERE to_archive = false
AND book.user_id = 1;

このSQLでは、bookテーブルからWHERE句の条件に該当する列を、userテーブルからusernameとemailを取り出して返却するようにしています。
SELECT文ではbookとuserのデータが結合されている前提で、取り出したい値を指定します。
肝はJOIN user ON book.user_id = user.idの部分で、JOINの後に結合したいテーブル名、ONの後に、何と何のデータを紐づけてテーブルを結合させるかの条件を指定します。
bookテーブルには、bookを登録したユーザーのidが保存されており、これはuserテーブル上のidと一致しているため、これに基づいてテーブルを連結させているわけです。

以上、解説してきたもので、だいたいSQLの基本構文を紹介することができたと思います。
次節からはMySQLとFastAPIを接続する方法を紹介しますが、MySQLの操作はORMの記法ではなく、生のSQL文を記述しています。特別な理由がなければORMの記法に合わせるべきですが、今回はSQLの学習も兼ねてアプリを作ろうとしていたため、生のSQL文を用いています。ご了承ください。

5-2. MySQLとFastAPIの接続

MySQLのデータベース用のコンテナーは、第三章のdocker composeの解説のところですでに登場しています。

docker-compose.yml
version: "3"

services:
  app:
  # ...

  db:
    image: mysql
    environment:
      MYSQL_DATABASE: ${MYSQL_DATABASE}
      MYSQL_USER: ${MYSQL_USER}
      MYSQL_PASSWORD: ${MYSQL_PASSWORD}
      MYSQL_ROOT_PASSWORD: ${MYSQL_ROOT_PASSWORD}
      TZ: "Asia/Tokyo"
    volumes:
      - mysql_data:/var/lib/mysql
    ports:
      - "3306:3306"

volumes:
  mysql_data:

※serviceと同じトップレベルでvolumeプロパティをしておかなければならないことに注意してください。詳細は第三章で解説しています。

さて、docker-compose.ymlファイルがある階層でdocker compose upコマンドを打てばデータベースがこれで立ち上がるわけですが、このままではFastAPIからMySQLコンテナーに接続することができません。

まず、sqlalchemyというORMライブラリーをインストールし、非同期処理の実装をサポートするaiomysqlもインストールします。
sqlalchemyやaiomysqlを用いることでMySQLのデータベースとの接続が確立され、テーブルを操作することが可能になります

FastAPI側での具体的な設定は以下のようなものです。

db.py
from sqlalchemy.engine.url import URL
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base
from os import environ

url = URL.create(
    drivername="mysql+aiomysql",
    username=environ.get('MYSQL_USER'),
    password=environ.get('MYSQL_PASSWORD'),
    host="db",
    database=environ.get('MYSQL_DATABASE'),
    query={"charset":"utf8"}
)

ASYNC_ENGINE = create_async_engine(
    url,
    echo=True
)

ASYNC_SESSION = sessionmaker(
            autocommit=False,
            autoflush=False,
            bind=ASYNC_ENGINE,
            class_=AsyncSession
        )

Base = declarative_base()

async def get_db():
    async with ASYNC_SESSION() as session:
        yield session

os.environ.get("")は、環境変数から値を取得してくるコードです。

ルーターにおいて、このget_db関数を用いてセッションを取得し、データベースにアクセスできるようになります。

コンテキストマネージャーであるwithを用いることで、処理が終了しブロックから抜けた場合、自動的にセッションが終了するようになります。
もしコンテキストマネジャーを用いない場合、例外処理（try-except-finally）を用いて、finallyブロックでセッションを閉じる処理（session.close）を書かなければなりません。

これで、データベースとの接続が完了しました。
しかし、まだデータベースのテーブルやその列のデータ型や制約などを設定できていないので、次節ではPythonの側からデータベースの初期設定を行う方法と、データベースとのデータのやり取りを行う方法を紹介します。

5-3. Modelとデータベースの操作

まず、データベースのモデル（テーブルの列やその型などの構造）を設定しなければなりません。
ここでは、bookモデルに絞って紹介します。

models/book.py
from sqlalchemy import Column, Integer, String, Boolean, DateTime ,ForeignKey
from datetime import datetime

from app.db import Base
from app.models.user import User

class Book(Base):
    __tablename__ = "book"

    id = Column(Integer, autoincrement=True ,primary_key=True)
    user_id = Column(Integer,ForeignKey(User.id))
    whole_pages = Column(Integer,nullable=False)
    read_pages = Column(Integer)
    created_at = Column(DateTime, default=datetime.now())
    deadline = Column(DateTime)
    title = Column(String(1024),nullable=False)
    comment = Column(String(1024))
    to_archive = Column(Boolean, default=False)

Bookモデルが継承しているBaseは、dbファイルで定義されたものです。

ここで作成したORMモデルを用いて、マイグレーションのための処理をmigrate_db.pyスクリプトに書きます。
以下の通りです。

migrate_db.py
from sqlalchemy import create_engine
from sqlalchemy.engine.url import URL
from app.models.user import Base as UserBase
from app.models.book import Base as BookBase
from app.models.archived_book import Base as ArchivedBookBase
from os import environ

url = URL.create(
    drivername="mysql+aiomysql",
    username=environ.get('MYSQL_USER'),
    password=environ.get('MYSQL_PASSWORD'),
    host="db",
    database=environ.get('MYSQL_DATABASE'),
    query={"charset":"utf8"}
)

ENGINE = create_engine(
    url,
    echo=True
)

def reset_database():
    UserBase.metadata.drop_all(bind=ENGINE)
    BookBase.metadata.drop_all(bind=ENGINE)
    ArchivedBookBase.metadata.drop_all(bind=ENGINE)

    UserBase.metadata.create_all(bind=ENGINE)
    BookBase.metadata.create_all(bind=ENGINE)
    ArchivedBookBase.metadata.create_all(bind=ENGINE)

if __name__ == "__main__":
    reset_database()

データベースに同名のテーブルがあった場合は、削除してから新規で作り直すようにしています。

docker compose exec app python -m app.migrate_dbコマンドを実行することで、migrate_dbファイルの処理を実行できます。

これでようやく、FastAPIがデータベースを操作する準備が整いました。
データベースの操作を記述したファイルは、crudsフォルダーにまとめてあります。
ここではbookテーブルに関わる記述に絞って紹介します。

cruds/book.py
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql import text
from sqlalchemy.engine.cursor import CursorResult
from typing import List, Optional
from logging import getLogger

from app.schemes import book as book_scheme
from app.schemes import user as user_scheme
from app.models import book as book_model
from app.models import archived_book as archived_book_model
from app.utils import error

logger = getLogger("uvicorn")

async def create_active_book(db: AsyncSession, book_create: book_scheme.CreateBook) -> book_scheme.ResponseBook:
    preparation = text("""
                            SELECT *
                            FROM book
                            WHERE :title in (SELECT DISTINCT title
                                            FROM book
                                            WHERE book.user_id = :user_id)
                            UNION
                            SELECT *
                            FROM archived_book
                            WHERE :title in (SELECT DISTINCT title
                                            FROM archived_book
                                            WHERE archived_book.user_id = :user_id)
                            LIMIT 1;
                        """)
    
    result: CursorResult = await db.execute(preparation,{"title":book_create.title, "user_id":book_create.user_id})

    if result.first() is not None:
        raise error.DuplicateError("The book has been already enrolled.")

    sql = text("""
                INSERT INTO book (user_id, whole_pages, read_pages, deadline, title, comment, to_archive)
                VALUES (:user_id, :whole_pages, :read_pages, :deadline, :title, :comment, :to_archive)
              """)

    await db.execute(sql, {
        "user_id": book_create.user_id,
        "whole_pages": book_create.whole_pages,
        "read_pages": book_create.read_pages,
        "deadline": book_create.deadline,
        "title": book_create.title,
        "comment": book_create.comment,
        "to_archive": book_create.to_archive
    })

    await db.commit()

    # 新たに登録された本のidを取得する
    result = await db.execute(text("SELECT LAST_INSERT_ID()"))
    book_id = result.scalar()

    result = await db.execute(text("SELECT email, username FROM user WHERE id = :id"), {"id": book_create.user_id})
    user_info = user_scheme.ConcatUserFromBook(**dict(zip(result.keys(), result.one())))

    book_info = dict(book_create)

    book_info.update({"id": book_id, "username": user_info.username, "email": user_info.email})

    return book_scheme.ResponseBook(**book_info)


async def get_active_books(db:AsyncSession, user_id:int) -> List[book_scheme.ResponseBook]:
    statement = text("""
                        SELECT book.*, user.username, user.email
                        FROM book
                        JOIN user
                        ON book.user_id = user.id
                        WHERE to_archive = false
                        AND book.user_id = :user_id;
                    """)

    result:CursorResult = await db.execute(statement,{"user_id":user_id})

    return [book_scheme.ResponseBook(**dict(zip(result.keys(),book_info))) for book_info in result.all()]


async def get_active_book(db:AsyncSession, book_id:int) -> book_scheme.ResponseBook:
    statement = text("""
                        SELECT book.*, user.username, user.email
                        FROM book
                        JOIN user
                        ON book.user_id = user.id
                        WHERE book.id = :id;
                    """)
    
    result:CursorResult = await db.execute(statement,{"id":book_id})

    book = result.first()
    if book is None:
        raise error.NoObjectError("There is no book you want.")

    return book_scheme.ResponseBook(**dict(zip(result.keys(),book)))

本来であれば、生のSQL文を送るようなことはせず、セッション管理機能を使ってデータベースとやり取りすることが望ましいですが、今回はSQLの練習のために生のSQL文を送るようにしています。

生のSQL文を送る際に注意したいのが、SELECT以外のsql命令文をデータベースに送ろうとするとその返り値が存在しないため、最新データを取得しなおさないといけないということです。
この課題は、MySQLに備わっているLAST_INSERT_ID()を用いて解決しています。

また、SELECT文を実行した時のデータは、データの配列でしかないため、このままでは期待していたようなオブジェクトの形になっていません。このため、テーブルの列情報を取得するkeys()メソッドを用いて列名とデータを結合し、データ型を満たすオブジェクトを作っています。

関数の返却部分でスキーマのクラスにオブジェクトを流し込んでいる理由は二つほどあります。
ひとつめは、スキーマクラスを取得データを引数にしてインスタンス化する過程においてプロパティの検証を行うため。
二つ目は、ルーターの受け取り側で返却されるデータの型が定められることで、プロパティアクセスがサポートされるためなどがあります。

ここで、もう一度ルーターの方を見てみましょう。
crudsの関数を取り込んだ形でパスオペレーションを実装しなおします。

routers/book_manager.py
from fastapi import APIRouter, Depends, HTTPException 
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List, Union

from app.schemes import book as book_scheme
from app.cruds import book as book_crud

router = APIRouter()

@router.get("/api/books",response_model=List[book_scheme.ResponseBook])
async def books(db:AsyncSession = Depends(get_db)):
    return await book_crud.get_active_books(db, user_id=1)

    
@router.post("/api/books",response_model=book_scheme.ResponseBook,status_code=201)
async def create_book(book_body: book_scheme.EnrollBook,response:Response, db: AsyncSession = Depends(get_db)):
    dict_book_body = dict(book_body)
    dict_book_body.update({"user_id":1})
    entire_book_info = book_scheme.CreateBook(**dict_book_body)
    
    try:
        book = await book_crud.create_active_book(db,entire_book_info)
    except error.DuplicateError:
        raise HTTPException(status_code=400,detail="The book has been already enrolled.")

    return book


@router.get("/api/books/{book_id}",response_model=book_scheme.ResponseBook)
async def book(book_id:int,db:AsyncSession = Depends(get_db)):
    try:
        book = await book_crud.get_active_book(db,book_id)
    except error.NoObjectError:
        raise HTTPException(status_code=404)

    return book

# ... patch delete　メソッドの定義

userのデータはJWT認証の部分で解説しますので、仮にuser_idは1にしておきます。
カスタムのエラーは、HTTExceptionエラーインスタンスを例外として投げることで実現します。
postメソッドにおいては、既にデータが登録されている場合は、重複エラーとして返すようにしています。
場合によっては、単に202 Acceptted (but not processed)を返すだけでも良いかもしれません。

一つ注意したいのは、依存性注入の部分です。
これは、コードが動作するために必要となる（依存関係にある）ものを宣言することです。
ロジックを共有している場合や、データベースとの接続が必要な場合、次章のJWT認証にも関わることですが、認証やセキュリティ機能を強制する時などに有用です。
上記の例を見てみると、パスオペレーションはdb:AsyncSession = Depends(get_db)としてセッションを受け取っています。get_dbはその名の通り、データベースとのセッションを提供する関数であったわけですから、dbの型注釈（AsyncSession）の通りというわけです。

最後に、JWT認証の機能を実装して、解説を終わろうと思います。

6. JWT認証

JWT認証に関するコードの紹介に先立って、インストールすべきライブラリーを二つ紹介します。

  • passlib
  • python-jose

passlibは、JWT認証の機能をFastAPIに組み込むにあたって、パスワードをハッシュ化するためのライブラリーです。
python-joseは、アクセストークンをつくったりトークンからデータを抽出するためのライブラリーです。

もう一度だけ、ディレクトー構成を見てみましょう。

C:.
│  .env
│  .gitignore
│  docker-compose.yml
│  Dockerfile
│  requirements.txt
│
└─app
    │  db.py
    │  main.py
    │  migrate_db.py
    │  __init__.py
    │
    ├─cruds
    │
    ├─models
    │
    ├─routers
    │
    ├─schemes
    │
    └─utils
       │  error.py
       │  jwt.py
       └─ pwd.py

JWT認証に関するファイルは、jwt.py, pwd.py, token.pyの三つのみです。

まず、token.pyの中身を紹介します。

schemes/token.py
from pydantic import BaseModel, EmailStr
from typing import Union

class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    email: Union[EmailStr, None] = None

このモジュールは、アクセストークンや、そこからメールアドレスのデータを取得する際に用いられるスキーマです。

次に、pwd.pyの紹介をします。

utils/pwd.py
from pydantic import SecretStr
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password:SecretStr, hashed_password:SecretStr):
    return pwd_context.verify(plain_password,hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

パスワードをハッシュ化する関数であるget_password_hashと、ログイン時に受け取ったパスワード（plain_password）をハッシュ化してデータベースから取り出したパスワード（hashed_password）を比較する関数であるverify_passwordを実装します。

JWT認証のコア機能は、jwt.pyスクリプトにおいて実装しています。

utils/jwt.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from pydantic import EmailStr
from jose import jwt, JWTError
from typing import Union
from datetime import datetime, timedelta

from app.schemes import user as user_scheme
from app.schemes import token as token_scheme
from app.cruds import auth as user_crud
from app.utils import pwd
from app.db import get_db

JWT_ALGORITHM = "HS256"
SECRET_KEY = "202043523ha34sh88_dammy4586keeyyd31"

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="api/auth/login")

credentials_exception = HTTPException(
    status_code=status.HTTP_401_UNAUTHORIZED,
    detail="Could not validate credentials",
)

# データベース上とやり取りする関数たち。
async def validate_user(email:EmailStr,db:AsyncSession):
    flag = await user_crud.validate_user(db,email)
    return flag


async def get_user(email:EmailStr,db:AsyncSession) -> user_scheme.ResponseUser :
    user = await user_crud.get_user(db,email)
    return user


async def authenticate_user(account_body: user_scheme.SignInUser,db:AsyncSession) -> Union[user_scheme.ResponseUser,False]:
    user = await user_crud.get_user(db,account_body.email)

    if not user or not pwd.verify_password(account_body.password, user.password):
        return False
    
    return user

# アクセストークンをつくったり取得したりする関数たち。
def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.now() + timedelta(hours=2)

    to_encode.update({"exp":expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=JWT_ALGORITHM)

    return encoded_jwt


def get_token(token:str = Depends(oauth2_scheme)) -> token_sheme.TokenData:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[JWT_ALGORITHM])
        email: str = payload.get("sub")

        if email is None:
            raise credentials_exception
        token_data = token_scheme.TokenData(email=email)

    except JWTError:
        raise credentials_exception
    
    return token_data

# userデータ（又はそのidのみ）を取得する関数たち
async def get_current_user(token_data:token_scheme.TokenData = Depends(get_token), db:AsyncSession = Depends(get_db)):
    user = await get_user(token_data.email,db)
    if not user:
        raise credentials_exception
    
    return user


async def get_current_user_id(token_data:token_scheme.TokenData = Depends(get_token), db:AsyncSession = Depends(get_db)):
    user = await get_user(token_data.email,db)
    if not user:
        raise credentials_exception
    
    return user.id

# アクセストークンが有効かどうかを検証する関数。
async def verify_token(token_data:token_scheme.TokenData = Depends(get_token), db:AsyncSession = Depends(get_db)):
    flag = await validate_user(token_data.email, db)
    if not flag:
        raise credentials_exception

JWTの仕様では、トークンのsubjectを表すキーsubがあるとされています。subにはユーザーの識別情報を含めるように定められているので、ここではemailを用いています。
ルーターのlogin処理の方でcreate_access_token関数にsubプロパティの値としてemailを渡しており、受け取ったアクセストークンを解析するget_token関数で、アクセストークンからメールアドレスを取り出しています。そしてこのemailによって、ユーザーの認証を行っているわけです。

最後に、認証機能まで実装したbookルーター機能を紹介して終わります。

routers/book_manager.py
from fastapi import APIRouter, Depends, HTTPException
from fastapi.security import HTTPBearer
from sqlalchemy.ext.asyncio import AsyncSession
from typing import List, Union

from app.cruds import book as book_crud
from app.schemes import book as book_scheme
from app.db import get_db
from app.utils import jwt

bearer_scheme = HTTPBearer()

router = APIRouter(
        prefix="/api/books",
        dependencies=[Depends(jwt.verify_token), Depends(bearer_scheme)],
        responses={401:{"description":"Not Authorized"}}
    )


@router.get("/",response_model=List[book_scheme.ResponseBook])
async def books(user_id:int = Depends(jwt.get_current_user_id) ,db:AsyncSession = Depends(get_db)):
    return await book_crud.get_active_books(db, user_id)


@router.post("/",response_model=book_scheme.ResponseBook,status_code=201)
async def create_book(book_body: book_scheme.EnrollBook, db: AsyncSession = Depends(get_db), user_id:int = Depends(jwt.get_current_user_id)):
    dict_book_body = dict(book_body)
    dict_book_body.update({"user_id":user_id})
    entire_book_info = book_scheme.CreateBook(**dict_book_body)
    try:
        book = await book_crud.create_active_book(db,entire_book_info)
    except error.DuplicateError:
        raise HTTPException(status_code=400,detail="The book has been already enrolled.")
    return book


@router.get("/{book_id}",response_model=book_scheme.ResponseBook)
async def book(book_id:int,db:AsyncSession = Depends(get_db)):
    try:
        book = await book_crud.get_active_book(db,book_id)
    except error.NoObjectError:
        raise HTTPException(status_code=404)
    return book

# ... patch delete メソッドの実装。

注目したいのは、APIRouterの引数として幾つかの引数を指定しているところです。これにより、そのパスに共通する処理を簡単に記述することができます。
prefixは、基底となるパスのことです。例えば、@router.get("/{book_id}")のパスデコレーターは、@router.get("/api/books/{book_id}")と同じ意味になります。
dependenciesは、依存性注入を配列で宣言することができます。この場合は、JWT認証に関わるオブジェクトを依存関係として指定しています。
responsesは、依存関係にある関数がエラーを出した場合などにどのようなレスポンスを返すかを定めます。この場合は、アクセストークンが不正であるということなので、401 Not Authorizedを返すようにしています。

これで、JWT認証のざっくりとした導入が終わりました。

7. おわりに

今回の記事では、個人開発の中で調べたことを包括的にすべて整理しようとしたのですが、過去一ながい記事になってしまいました。
自分のコードを見返すときにこの記事を読みながらであれば、一年間プログラミングから離れていても分かるくらいには丁寧にまとめられたかなって感じです。
エラーと格闘しながら開発を進めたのも其々の要素の深い理解につながっているので、とてもためになる経験でした。

しかし、APIのアップデートのノウハウやセキュリティの話、SQLのより高度な構文など、まだまだ調べたりないところはあります。
これからも継続して学習を進め、開発を通じて体験し、そしてそれをこのような形で整理してアウトプットするサイクルを続けていきたいと思います。
ここまで読んでくださり、ありがとうございました。

8. 参考

第三章の参考資料

第四章の参考資料

第五章の参考資料

第六章の参考資料

  1. 他にもcontainer commitでコンテナーを直接イメージに変換したり、container export を用いてコンテナーをtarアーカイブファイルに変換してからイメージをつくったりする方法もあります。 ↩

1
3
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
Sign upLogin
1
3