FastAPIを使ってメモアプリを作ってみた

## TL;DR
前回はFlaskでMPA（Multi Page Application）を書きましたが、今回はFastAPIを用いてSPA（Single Page Application）のバックエンドを書きつつ、将来DDD（Domain Driven Development）をするため、どのようなことを意識したらいいのか考えてみました。

以下のリポジトリにサンプルコードを公開しました。フォルダ構成や記事の中で出てこないファイルなどはこちらをご覧ください。

サンプルコード→ https://github.com/hiseumn/sample-memo-fastapi-backend

また、参考資料に前回の記事を掲載しています。一通り動くところまでの手順はそちらに書いたため、今回は省略します。

## 必要なライブラリのインストール
今回もパッケージマネージャーはRyeを使用しています。

### SQLAlchemy
今回はSQL文を直接書かず、ORMを使用しました。
```zsh:terminal
rye add sqlalchemy
```

### FastAPI
SPAのフロントエンドとの連携には通常REST APIでの連携となり、パラメーターやレスポンスにはJSON形式が用いられます。このためJSONオブジェクトをPythonのクラスオブジェクトに変換したり、バリデーションをしてくれるPydanticとの連携が前提となっているFastAPIを用います。Flaskでも同様のことができますが、高速なパフォーマンス・豊富な自動生成機能・非同期処理のサポートなど様々なメリットを享受できるため、FastAPIを用います。
```zsh:terminal
rye add fastapi
```

### uuid
保存するメモのIDとしてUUIDを使用しました。IDを時系列でソートしたかったので、ビルトインのv4ではなくv7を使用するため、以下のライブラリをインストールしています。
```zsh:terminal
rye add uuid6
```

## 主なファイル構成

フォルダの構成は以下のとおりです。
```
src
    └── sample_fastapi_memo
        ├── app.py
        ├── database.py
        ├── functions.py
        ├── routers.py
        └── schemas.py
```

### app.py
routerをincludeしているだけですが、アプリケーションが対応するドメインが増えたら、ここでそのドメインへのrouterを追記していきます。
```python:python
from routers import router
from fastapi import FastAPI

app = FastAPI()
app.include_router(router)

```

### database.py
DDDで言うインフラ層に相当します。PostgreSQLとのやりとりはここにまとめます。
```python:python
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.schema import Column
from uuid6 import uuid7
from sqlalchemy_utils import UUIDType
from sqlalchemy.types import Text

Base = declarative_base()

engine=create_engine(f"postgresql+psycopg://postgres:postgres@localhost:5432/memo")

def get_db_session() -> scoped_session:
    """ 新しいDBコネクションを返す
    """
    return scoped_session(sessionmaker(bind=engine))

class Memo(Base):
    __tablename__ = "memo"  # テーブル名を指定
    id = Column(UUIDType(binary=False), primary_key=True, default=uuid7)
    memo = Column(Text)

    def memo_result(self):  # メモの内容を返す
        return "{self.id} {self.memo}"
    
```
※データベースに接続するためのIDとパスワードをハードコーディングすることは、セキュリティリスクになりかねないため真似しないでください。

### functions.py
今回はあまり厳密に分けていませんが、DDDで言うユースケースとドメインの部分に相当します。このアプリの実際のロジックをここにまとめます。
```python:python
from sqlalchemy import desc
from sqlalchemy.orm import scoped_session
from database import Memo
from schemas import InputMemo as InputMemoSchema, Memo as MemoSchema
from uuid6 import uuid7

def add_memo(memo: InputMemoSchema, session: scoped_session) -> Memo | None:
    memo = Memo(id=uuid7(), memo=memo.memo)
    session.add(memo)
    session.commit()
    session.refresh(memo)
    return memo


def get_memo(session: scoped_session):
    return session.query(Memo).order_by(desc('id')).limit(20).all()

```

### routers.py
APIのエンドポイントを記述します。ここではビジネスロジックはできる限り書かないようにしています。
```python:python
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import scoped_session
import functions
from database import get_db_session
from schemas import InputMemo, Memo

router = APIRouter()


@router.get("/memo", tags=["/memo"])
def get_memo(session: scoped_session = Depends(get_db_session)) -> list[Memo]:
    memo = functions.get_memo(session)
    return list(map(Memo.model_validate, memo))


@router.post("/memo", tags=["/memo"])
def add_memo(memo: InputMemo, session: scoped_session = Depends(get_db_session)) -> Memo:
    memo = functions.add_memo(memo, session)
    return Memo.model_validate(memo)

```

### schemas.py
似たようなクラスが`database.py`にもありますが、こちらはAPIのエンドポイントに関するリクエストとレスポンスのパラメーターを定義しています。`InputMemo`と`Memo`に分けたのは、PostのパラメーターとGetのレスポンスでは同じMemoでも必須となる項目が違うからです。
```python:python
from uuid import UUID
from pydantic import BaseModel, ConfigDict


class InputMemo(BaseModel):
    memo: str
    model_config = ConfigDict(from_attributes=True)


class Memo(InputMemo):
    id: UUID

```

##　参考資料

https://qiita.com/hiseumn/items/0ced9aa4a1f81a797d26

https://qiita.com/hiseumn/items/5992c6818a18210f84f8

以上