fastapi-best-practices - @ hi.peerlink.me で使用した FastAPI のベスト プラクティスと規則

(FastAPI Best Practices and Conventions we used @ hi.peerlink.me)

Created at: 2022-08-09 14:03:38
Language:

FastAPI のベスト プラクティス

スタートアップで使用したベスト プラクティスと規則の独自のリスト。

本番環境での過去 1.5 年間、開発者エクスペリエンスに劇的な影響を与える良い決定と悪い決定を下してきました。それらのいくつかは共有する価値があります。

コンテンツ

  1. プロジェクト構造。一貫性があり、予測可能です。
  2. データ検証に Pydantic を過度に使用します。
  3. データ検証と DB の依存関係を使用します。
  4. チェーンの依存関係。
  5. 依存関係を切り離して再利用します。依存関係の呼び出しはキャッシュされます。
  6. RESTに従ってください。
  7. I/O 操作をブロックするだけの場合は、ルートを非同期にしないでください。
  8. Day 0 からのカスタム ベース モデル。
  9. ドキュメント。
  10. 設定には Pydantic の BaseSettings を使用します。
  11. SQLAlchemy: DB キーの命名規則を設定します。
  12. 移行。アレンビック。
  13. DB命名規則を設定します。
  14. テスト クライアントを 0 日目から非同期に設定します。
  15. BackgroundTasks > asyncio.create_task.
  16. タイピングは重要です。
  17. ファイルをチャンクで保存します。
  18. 動的ピダンティック フィールドには注意してください。
  19. SQL ファースト、Pydantic セカンド。
  20. ユーザーが公開されている URL を送信できる場合は、ホストを検証します。
  21. スキーマが直接クライアントに面している場合、カスタム pydantic バリデータで ValueError を発生させます。
  22. FastAPI が Response Pydantic Object を変換することを忘れないでください...
  23. 同期 SDK を使用する必要がある場合は、スレッド プールで実行します。
  24. リンター (黒、アイソルト、オートフレーク) を使用します。
  25. ボーナスセクション。

1. プロジェクト構造。一貫性と予測可能性

プロジェクトを構成する方法はたくさんありますが、最適な構成は、一貫性があり、単純明快で、驚きのない構成です。

  • プロジェクトの構造を見てもプロジェクトの内容がわからない場合は、構造が不明確である可能性があります。
  • どのモジュールが含まれているかを理解するためにパッケージを開く必要がある場合、構造が不明確になります。
  • ファイルの頻度と場所がランダムに感じられる場合は、プロジェクトの構造が正しくありません。
  • モジュールの場所とその名前を見て、その中に何が入っているかがわからない場合は、構造が非常に悪いです。

@tiangoloによって提示されたタイプ (例: api、crud、モデル、スキーマ) ごとにファイルを分けたプロジェクト構造は、スコープの少ないマイクロサービスやプロジェクトには適していますが、多数のドメインを持つモノリスには適合しませんでした。およびモジュール。よりスケーラブルで進化可能な構造は、Netflix のDispatchに少し変更を加えたものに触発されています。

fastapi-project
├── alembic/
├── src
│   ├── auth
│   │   ├── router.py
│   │   ├── schemas.py  # pydantic models
│   │   ├── models.py  # db models
│   │   ├── dependencies.py
│   │   ├── config.py  # local configs
│   │   ├── constants.py
│   │   ├── exceptions.py
│   │   ├── service.py
│   │   └── utils.py
│   ├── aws
│   │   ├── client.py  # client model for external service communication
│   │   ├── schemas.py
│   │   ├── config.py
│   │   ├── constants.py
│   │   ├── exceptions.py
│   │   └── utils.py
│   └── posts
│   │   ├── router.py
│   │   ├── schemas.py
│   │   ├── models.py
│   │   ├── dependencies.py
│   │   ├── constants.py
│   │   ├── exceptions.py
│   │   ├── service.py
│   │   └── utils.py
│   ├── config.py  # global configs
│   ├── models.py  # global models
│   ├── exceptions.py  # global exceptions
│   ├── pagination.py  # global module e.g. pagination
│   ├── database.py  # db connection related stuff
│   └── main.py
├── tests/
│   ├── auth
│   ├── aws
│   └── posts
├── templates/
│   └── index.html
├── requirements
│   ├── base.txt
│   ├── dev.txt
│   └── prod.txt
├── .env
├── .gitignore
├── logging.ini
└── alembic.ini
  1. すべてのドメイン ディレクトリを
    src
    フォルダー 内に格納する
    1. src/
      - アプリの最高レベル。共通のモデル、構成、定数などが含まれます。
    2. src/main.py
      - FastAPI アプリを初期化するプロジェクトのルート
  2. 各パッケージには、独自のルーター、スキーマ、モデルなどがあります。
    1. router.py
      - すべてのエンドポイントを持つ各モジュールのコアです
    2. schemas.py
      - ピダンティック モデル用
    3. models.py
      - db モデルの場合
    4. service.py
      - モジュール固有のビジネス ロジック
    5. dependencies.py
      - ルーターの依存関係
    6. constants.py
      - モジュール固有の定数とエラー コード
    7. config.py
      - 例: 環境変数
    8. utils.py
      - ビジネス ロジック以外の機能 (応答の正規化、データの強化など)。
    9. exceptions
      - モジュール固有の例外
      PostNotFound
      InvalidUserData
  3. パッケージが他のパッケージからのサービスまたは依存関係または定数を必要とする場合 - それらを明示的なモジュール名でインポートします
from src.auth import constants as auth_constants
from src.notifications import service as notification_service
from src.posts.constants import ErrorCode as PostsErrorCode  # in case we have Standard ErrorCode in constants module of each package

2. データ検証に Pydantic を過度に使用する

Pydantic には、データを検証および変換するための豊富な機能セットがあります。

デフォルト値を持つ必須および必須でないフィールドなどの通常の機能に加えて、Pydantic には、正規表現、許可された限られたオプションの列挙、長さの検証、電子メールの検証などの包括的なデータ処理ツールが組み込まれています。

from enum import Enum
from pydantic import AnyUrl, BaseModel, EmailStr, Field, constr

class MusicBand(str, Enum):
   AEROSMITH = "AEROSMITH"
   QUEEN = "QUEEN"
   ACDC = "AC/DC"


class UserBase(BaseModel):
    first_name: str = Field(min_length=1, max_length=128)
    username: constr(regex="^[A-Za-z0-9-_]+$", to_lower=True, strip_whitespace=True)
    email: EmailStr
    age: int = Field(ge=18, default=None)  # must be greater or equal to 18
    favorite_band: MusicBand = None  # only "AEROSMITH", "QUEEN", "AC/DC" values are allowed to be inputted
    website: AnyUrl = None

3. データ検証と DB の依存関係を使用する

Pydantic は、クライアント入力の値のみを検証できます。依存関係を使用して、電子メールが既に存在する、ユーザーが見つからないなどのデータベースの制約に対してデータを検証します。

# dependencies.py
async def valid_post_id(post_id: UUID4) -> Mapping:
    post = await service.get_by_id(post_id)
    if not post:
        raise PostNotFound()

    return post


# router.py
@router.get("/posts/{post_id}", response_model=PostResponse)
async def get_post_by_id(post: Mapping = Depends(valid_post_id)):
    return post


@router.put("/posts/{post_id}", response_model=PostResponse)
async def update_post(
    update_data: PostUpdate,  
    post: Mapping = Depends(valid_post_id), 
):
    updated_post: Mapping = await service.update(id=post["id"], data=update_data)
    return updated_post


@router.get("/posts/{post_id}/reviews", response_model=list[ReviewsResponse])
async def get_post_reviews(post: Mapping = Depends(valid_post_id)):
    post_reviews: list[Mapping] = await reviews_service.get_by_post_id(post["id"])
    return post_reviews

データ検証を依存関係に入れなかった場合、すべてのエンドポイントに post_id 検証を追加し、それぞれに同じテストを作成する必要があります。

4. チェーンの依存関係

依存関係は他の依存関係を使用して、同様のロジックのコードの繰り返しを避けることができます。

# dependencies.py
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt

async def valid_post_id(post_id: UUID4) -> Mapping:
    post = await service.get_by_id(post_id)
    if not post:
        raise PostNotFound()

    return post


async def parse_jwt_data(
    token: str = Depends(OAuth2PasswordBearer(tokenUrl="/auth/token"))
) -> dict:
    try:
        payload = jwt.decode(token, "JWT_SECRET", algorithms=["HS256"])
    except JWTError:
        raise InvalidCredentials()

    return {"user_id": payload["id"]}


async def valid_owned_post(
    post: Mapping = Depends(valid_post_id), 
    token_data: dict = Depends(parse_jwt_data),
) -> Mapping:
    if post["creator_id"] != token_data["user_id"]:
        raise UserNotOwner()

    return post

# router.py
@router.get("/users/{user_id}/posts/{post_id}", response_model=PostResponse)
async def get_user_post(post: Mapping = Depends(valid_owned_post)):
    """Get post that belong the user."""
    return post

5. 依存関係を切り離して再利用する。依存関係の呼び出しはキャッシュされます。

依存関係は複数回再利用でき、再計算されることはありません。FastAPI はデフォルトで依存関係の結果をリクエストのスコープ内にキャッシュします。つまり、 service を呼び出す依存関係がある場合、

get_post_by_id
この依存関係を呼び出すたびに DB にアクセスすることはありません。最初の関数呼び出しのみ。

これを知っていれば、より小さなドメインで動作し、他のルートで再利用しやすい複数の小さな関数への依存関係を簡単に切り離すことができます。たとえば、以下のコードでは、

parse_jwt_data
3 回使用しています。

  1. valid_owned_post
  2. valid_active_creator
  3. get_user_post

ただし

parse_jwt_data
、最初の呼び出しで一度だけ呼び出されます。

# dependencies.py
from fastapi import BackgroundTasks
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt

async def valid_post_id(post_id: UUID4) -> Mapping:
    post = await service.get_by_id(post_id)
    if not post:
        raise PostNotFound()

    return post


async def parse_jwt_data(
    token: str = Depends(OAuth2PasswordBearer(tokenUrl="/auth/token"))
) -> dict:
    try:
        payload = jwt.decode(token, "JWT_SECRET", algorithms=["HS256"])
    except JWTError:
        raise InvalidCredentials()

    return {"user_id": payload["id"]}


async def valid_owned_post(
    post: Mapping = Depends(valid_post_id), 
    token_data: dict = Depends(parse_jwt_data),
) -> Mapping:
    if post["creator_id"] != token_data["user_id"]:
        raise UserNotOwner()

    return post


async def valid_active_creator(
    token_data: dict = Depends(parse_jwt_data),
):
    user = await users_service.get_by_id(token_data["user_id"])
    if not user["is_active"]:
        raise UserIsBanned()
    
    if not user["is_creator"]:
       raise UserNotCreator()
    
    return user
        

# router.py
@router.get("/users/{user_id}/posts/{post_id}", response_model=PostResponse)
async def get_user_post(
    worker: BackgroundTasks,
    post: Mapping = Depends(valid_owned_post),
    user: Mapping = Depends(valid_active_creator),
):
    """Get post that belong the active user."""
    worker.add_task(notifications_service.send_email, user["id"])
    return post

6.RESTに従う

RESTful API を開発すると、次のようなルートで依存関係を再利用しやすくなります。

  1. GET /courses/:course_id
  2. GET /courses/:course_id/chapters/:chapter_id/lessons
  3. GET /chapters/:chapter_id

唯一の注意点は、パスで同じ変数名を使用することです。

  • 2 つのエンドポイントが
    GET /profiles/:profile_id
    あり
    GET /creators/:creator_id
    、両方が指定されたもの
    profile_id
    が存在するかどうかを検証するだけでなく
    GET /creators/:creator_id
    、プロファイルが作成者かどうかもチェックする場合は、
    creator_id
    パス変数の名前を変更して
    profile_id
    、これら 2 つの依存関係をチェーンすることをお勧めします。
# src.profiles.dependencies
async def valid_profile_id(profile_id: UUID4) -> Mapping:
    profile = await service.get_by_id(post_id)
    if not profile:
        raise ProfileNotFound()

    return profile

# src.creators.dependencies
async def valid_creator_id(profile: Mapping = Depends(valid_profile_id)) -> Mapping:
    if not profile["is_creator"]:
       raise ProfileNotCreator()

    return profile

# src.profiles.router.py
@router.get("/profiles/{profile_id}", response_model=ProfileResponse)
async def get_user_profile_by_id(profile: Mapping = Depends(valid_profile_id)):
    """Get profile by id."""
    return profile

# src.creators.router.py
@router.get("/creators/{profile_id}", response_model=ProfileResponse)
async def get_user_profile_by_id(
     creator_profile: Mapping = Depends(valid_creator_id)
):
    """Get profile by id."""
    return creator_profile

ユーザー自身のリソースに /me エンドポイントを使用する (例

GET /profiles/me
:
GET /users/me/posts
)

  1. ユーザー ID が存在することを検証する必要はありません - 認証方法によって既にチェックされています
  2. ユーザーIDがリクエスタに属しているかどうかを確認する必要はありません

7. I/O 操作をブロックするだけの場合は、ルートを非同期にしないでください

内部では、FastAPI は非同期と同期の両方の I/O 操作を効果的に処理できます。

  • FastAPI はスレッドプールでルートを実行
    sync
    、 I/O 操作をブロックしても、イベント ループ によるタスクの実行は停止しません。
  • それ以外の場合、ルートが定義されている場合は
    async
    定期的に呼び出され
    await
    、FastAPI は非ブロッキング I/O 操作のみを実行することを信頼します。

注意点として、信頼に失敗し、非同期ルート内でブロック操作を実行すると、そのブロック操作が完了するまで、イベント ループは次のタスクを実行できなくなります。

import asyncio
import time

@router.get("/terrible-ping")
async def terrible_catastrophic_ping():
    time.sleep(10) # I/O blocking operation for 10 seconds
    pong = service.get_pong()  # I/O blocking operation to get pong from DB
    
    return {"pong": pong}

@router.get("/good-ping")
def good_ping():
    time.sleep(10) # I/O blocking operation for 10 seconds, but in another thread
    pong = service.get_pong()  # I/O blocking operation to get pong from DB, but in another thread
    
    return {"pong": pong}

@router.get("/perfect-ping")
async def perfect_ping():
    await asyncio.sleep(10) # non I/O blocking operation
    pong = await service.async_get_pong()  # non I/O blocking db call

    return {"pong": pong}

呼び出すとどうなりますか:

  1. GET /terrible-ping
    1. FastAPI サーバーがリクエストを受け取り、処理を開始します
    2. time.sleep()
      サーバーのイベント ループとキュー内のすべてのタスクが終了 するまで待機します。
      1. サーバーは
        time.sleep()
        I/O タスクではないと判断するため、完了するまで待機します
      2. 待機中、サーバーは新しいリクエストを受け入れません
    3. service.get_pong
      次に、イベントループとキュー内のすべてのタスクが終了 するまで待機します
      1. サーバーは
        service.get_pong()
        I/O タスクではないと判断するため、完了するまで待機します
      2. 待機中、サーバーは新しいリクエストを受け入れません
    4. サーバーが応答を返します。
      1. 応答後、サーバーは新しい要求の受け入れを開始します
  2. GET /good-ping
    1. FastAPI サーバーがリクエストを受け取り、処理を開始します
    2. FastAPI
      good_ping
      は、ワーカー スレッドが関数を実行するスレッドプールにルート全体を送信します。
    3. が実行されている間
      good_ping
      、イベント ループはキューから次のタスクを選択して処理します (たとえば、新しい要求を受け入れ、db を呼び出します)。
      • メイン スレッド (つまり、FastAPI アプリ) とは無関係に、ワーカー スレッドは終了するのを待ってから
        time.sleep
        終了するのを待ちます。
        service.get_pong
    4. good_ping
      作業が完了すると、サーバーはクライアントに応答を返します
  3. GET /perfect-ping
    1. FastAPI サーバーがリクエストを受け取り、処理を開始します
    2. FastAPI が待っています
      asyncio.sleep(10)
    3. イベント ループは、キューから次のタスクを選択して処理します (たとえば、新しい要求を受け入れ、db を呼び出します)。
    4. 完了する
      asyncio.sleep(10)
      と、サーバーは次の行に移動して待機します
      service.async_get_pong
    5. イベント ループは、キューから次のタスクを選択して処理します (たとえば、新しい要求を受け入れ、db を呼び出します)。
    6. 完了する
      service.async_get_pong
      と、サーバーはクライアントに応答を返します

2 つ目の注意点は、非ブロッキングの待機可能操作またはスレッド プールに送信される操作は、I/O 集中型タスク (ファイルを開く、データベース呼び出し、外部 API 呼び出しなど) である必要があることです。

  • CPU を集中的に使用するタスク (重い計算、データ処理、ビデオ トランスコーディングなど) を待機しても意味がありません。なぜなら、CPU はタスクを完了するために動作する必要があるためです。一方、I/O 操作は外部で行われ、その操作が完了するのを待っている間、サーバーは何もしません。次のタスクに進むことができます。
  • GILのため、他のスレッドで CPU を集中的に使用するタスクを実行することも効果的ではありません。つまり、GIL では一度に 1 つのスレッドしか動作できないため、CPU タスクには役に立ちません。
  • CPU を集中的に使用するタスクを最適化したい場合は、それらを別のプロセスのワーカーに送信する必要があります。

混乱したユーザーの関連する StackOverflow の質問

  1. https://stackoverflow.com/questions/62976648/architecture-flask-vs-fastapi/70309597#70309597
  2. https://stackoverflow.com/questions/65342833/fastapi-uploadfile-is-slow-compared-to-flask
  3. https://stackoverflow.com/questions/71516140/fastapi-runs-api-calls-in-serial-instead-of-parallel-fashion

8. 0 日目からのカスタム ベース モデル。

制御可能なグローバル ベース モデルを使用すると、アプリ内のすべてのモデルをカスタマイズできます。たとえば、標準の日時形式を使用したり、基本モデルのすべてのサブクラスにスーパー メソッドを追加したりできます。

from datetime import datetime
from zoneinfo import ZoneInfo

import orjson
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel, root_validator


def orjson_dumps(v, *, default):
    # orjson.dumps returns bytes, to match standard json.dumps we need to decode
    return orjson.dumps(v, default=default).decode()


def convert_datetime_to_gmt(dt: datetime) -> str:
    if not dt.tzinfo:
        dt = dt.replace(tzinfo=ZoneInfo("UTC"))

    return dt.strftime("%Y-%m-%dT%H:%M:%S%z")


class ORJSONModel(BaseModel):
    class Config:
        json_loads = orjson.loads
        json_dumps = orjson_dumps
        json_encoders = {datetime: convert_datetime_to_gmt}  # method for customer JSON encoding of datetime fields

    @root_validator()
    def set_null_microseconds(cls, data: dict) -> dict:
       """Drops microseconds in all the datetime field values."""
        datetime_fields = {
            k: v.replace(microsecond=0)
            for k, v in data.items()
            if isinstance(k, datetime)
        }

        return {**data, **datetime_fields}

    def serializable_dict(self, **kwargs):
       """Return a dict which contains only serializable fields."""
        default_dict = super().dict(**kwargs)

        return jsonable_encoder(default_dict)

上記の例では、次のようなグローバル ベース モデルを作成することにしました。

  • orjsonを使用してデータをシリアル化する
  • すべての日付形式でマイクロ秒を 0 に落とします
  • 明示的なタイムゾーンを使用して、すべての日時フィールドを標準形式にシリアル化します

9. ドキュメント

  1. API が公開されていない限り、デフォルトでドキュメントを非表示にします。選択した環境でのみ明示的に表示します。
from fastapi import FastAPI
from starlette.config import Config

config = Config(".env")  # parse .env file for env variables

ENVIRONMENT = config("ENVIRONMENT")  # get current env name
SHOW_DOCS_ENVIRONMENT = ("local", "staging")  # explicit list of allowed envs

app_configs = {"title": "My Cool API"}
if ENVIRONMENT not in SHOW_DOCS_ENVIRONMENT:
   app_configs["openapi_url"] = None  # set url for docs as null

app = FastAPI(**app_configs)
  1. FastAPI がわかりやすいドキュメントを生成するのに役立ちます
    1. response_model
      status_code
      description
      などを設定します。
    2. モデルとステータスが異なる場合は、
      responses
      ルート属性を使用して、さまざまな応答のドキュメントを追加します
from fastapi import APIRouter, status

router = APIRouter()

@router.post(
    "/endpoints",
    response_model=DefaultResponseModel,  # default response pydantic model 
    status_code=status.HTTP_201_CREATED,  # default status code
    description="Description of the well documented endpoint",
    tags=["Endpoint Category"],
    summary="Summary of the Endpoint",
    responses={
        status.HTTP_200_OK: {
            "model": OkResponse, # custom pydantic model for 200 response
            "description": "Ok Response",
        },
        status.HTTP_201_CREATED: {
            "model": CreatedResponse,  # custom pydantic model for 201 response
            "description": "Creates something from user request ",
        },
        status.HTTP_202_ACCEPTED: {
            "model": AcceptedResponse,  # custom pydantic model for 202 response
            "description": "Accepts request and handles it later",
        },
    },
)
async def documented_route():
    pass

次のようなドキュメントを生成します。 FastAPI によって生成されたカスタム レスポンス ドキュメント

10. 構成に Pydantic の BaseSettings を使用する

Pydantic は、環境変数を解析し、バリデーターで処理するための強力なツールを提供します。

from pydantic import AnyUrl, BaseSettings, PostgresDsn

class AppSettings(BaseSettings):
    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"
        env_prefix = "app_"

    DATABASE_URL: PostgresDsn
    IS_GOOD_ENV: bool = True
    ALLOWED_CORS_ORIGINS: set[AnyUrl]

11. SQLAlchemy: DB キーの命名規則を設定する

データベースの規則に従ってインデックスの命名を明示的に設定することは、sqlalchemy の規則よりも望ましいことです。

from sqlalchemy import MetaData

POSTGRES_INDEXES_NAMING_CONVENTION = {
    "ix": "%(column_0_label)s_idx",
    "uq": "%(table_name)s_%(column_0_name)s_key",
    "ck": "%(table_name)s_%(constraint_name)s_check",
    "fk": "%(table_name)s_%(column_0_name)s_fkey",
    "pk": "%(table_name)s_pkey",
}
metadata = MetaData(naming_convention=POSTGRES_INDEXES_NAMING_CONVENTION)

12. 移行。アレンビック。

  1. 移行は静的で元に戻せる必要があります。移行が動的に生成されたデータに依存している場合は、動的なのは構造ではなく、データ自体だけであることを確認してください。
  2. わかりやすい名前とスラッグで移行を生成します。スラッグは必須であり、変更を説明する必要があります。
  3. 新しい移行用に人間が判読できるファイル テンプレートを設定します。
    *date*_*slug*.py
    パターンを使用します。
    2022-08-24_post_content_idx.py
# alembic.ini
file_template = %%(year)d-%%(month).2d-%%(day).2d_%%(slug)s

13.DBの命名規則を設定する

名前に一貫性を持たせることは重要です。私たちが従ったいくつかのルール:

  1. lower_case_snake
  2. 単数形 (例:
    post
    ,
    post_like
    ,
    user_playlist
    )
  3. 、、、などのモジュール プレフィックスを使用して、類似したテーブルをグループ化し
    payment_account
    ます。
    payment_bill
    post
    post_like
  4. テーブル全体で一貫性を保ちますが、具体的な命名は問題ありません。
    1. profile_id
      すべてのテーブルで使用しますが、そのうちのいくつかが作成者であるプロファイルのみを必要とする場合は、使用します
      creator_id
    2. post_id
      のようなすべての抽象テーブルに使用しますが、関連するモジュールでは具体的な命名を使用します
      post_like
      post_view
      course_id
      chapters.course_id
  5. _at
    日時のサフィックス
  6. _date
    日付のサフィックス

14. テスト クライアントを 0 日目から非同期に設定する

DB との統合テストを作成すると、将来、めちゃくちゃなイベント ループ エラーが発生する可能性が高くなります。async_asgi_testclienthttpxなど、非同期テスト クライアントをすぐに設定します。

import pytest
from async_asgi_testclient import TestClient

from src.main import app  # inited FastAPI app


@pytest.fixture
async def client():
    host, port = "127.0.0.1", "5555"
    scope = {"client": (host, port)}

    async with TestClient(
        app, scope=scope, headers={"X-User-Fingerprint": "Test"}
    ) as client:
        yield client


@pytest.mark.asyncio
async def test_create_post(client: TestClient):
    resp = await client.post("/posts")

    assert resp.status_code == 201

db 接続を同期している場合 (すみません?)、または統合テストを作成する予定がない場合を除きます。

15. BackgroundTasks > asyncio.create_task

BackgroundTasks は 、ルートを処理するのと同じ方法で、ブロッキング I/O 操作と非ブロッキング I/O 操作の両方を効果的に実行

sync
できます (関数はスレッドプールで実行され、関数は
async
後で待機します) 。

  • ワーカーに嘘をつかないでください。ブロックしている I/O 操作を次のようにマークしないでください。
    async
  • CPU を集中的に使用するタスクには使用しないでください。
from fastapi import APIRouter, BackgroundTasks
from pydantic import UUID4

from src.notifications import service as notifications_service


router = APIRouter()


@router.post("/users/{user_id}/email")
async def send_user_email(worker: BackgroundTasks, user_id: UUID4):
    """Send email to user"""
    worker.add_task(notifications_service.send_email, user_id)  # send email after responding client
    return {"status": "ok"}

16.タイピングは重要

FastAPI、Pydantic、および最新の IDE では、型ヒントの使用が推奨されています。

型ヒントなし

型ヒント付き

17. ファイルをチャンクに保存します。

クライアントが小さなファイルを送信することを期待しないでください。

import aiofiles
from fastapi import UploadFile

DEFAULT_CHUNK_SIZE = 1024 * 1024 * 50  # 50 megabytes

async def save_video(video_file: UploadFile):
   async with aiofiles.open("/file/path/name.mp4", "wb") as f:
     while chunk := await video_file.read(DEFAULT_CHUNK_SIZE):
         await f.write(chunk)

18. 動的ピダンティック フィールドに注意する

型の共用体を受け入れることができる pydantic フィールドがある場合は、バリデーターがそれらの型の違いを明示的に認識していることを確認してください。

from pydantic import BaseModel


class Article(BaseModel):
   text: str | None
   extra: str | None


class Video(BaseModel):
   video_id: int
   text: str | None
   extra: str | None

   
class Post(BaseModel):
   content: Article | Video

   
post = Post(content={"video_id": 1, "text": "text"})
print(type(post.content))
# OUTPUT: Article
# Article is very inclusive and all fields are optional, allowing any dict to become valid

ひどくない解決策:

  1. フィールド タイプを適切に並べ替えます: 最も厳密なものから緩いものまで。
class Post(BaseModel):
   content: Video | Article
  1. 入力に有効なフィールドのみがあることを検証する
from pydantic import BaseModel, root_validator

class Article(BaseModel):
   text: str | None
   extra: str | None
   
   @root_validator(pre=True)  # validate all values before pydantic
   def has_only_article_fields(cls, data: dict):
      """Silly and ugly solution to validate data has only article fields."""
      fields = set(data.keys())
      if fields != {"text", "extra"}:
         raise ValueError("invalid fields")

      return data
       

class Video(BaseModel):
   video_id: int
   text: str | None
   extra: str | None
   
   @root_validator(pre=True)
   def has_only_video_fields(cls, data: dict):
      """Silly and ugly solution to validate data has only video fields."""
      fields = set(data.keys())
      if fields != {"text", "extra", "video_id"}:
         raise ValueError("invalid fields")

      return data

   
class Post(BaseModel):
   content: Article | Video
  1. フィールドが単純な場合は、Pydantic の Smart Union (>v1.9) を使用します。

int
フィールドがやのように単純な場合は良い解決策ですが、
bool
クラスのような複雑なフィールドでは機能しません。

Without Smart Union

from pydantic import BaseModel


class Post(BaseModel):
   field_1: bool | int
   field_2: int | str
   content: Article | Video

p = Post(field_1=1, field_2="1", content={"video_id": 1})
print(p.field_1)
# OUTPUT: True
print(type(p.field_2))
# OUTPUT: int
print(type(p.content))
# OUTPUT: Article

With Smart Union

class Post(BaseModel):
   field_1: bool | int
   field_2: int | str
   content: Article | Video

   class Config:
      smart_union = True


p = Post(field_1=1, field_2="1", content={"video_id": 1})
print(p.field_1)
# OUTPUT: 1
print(type(p.field_2))
# OUTPUT: str
print(type(p.content))
# OUTPUT: Article, because smart_union doesn't work for complex fields like classes

19. SQL-first, Pydantic-second

  • Usually, database handles data processing much faster and cleaner than CPython will ever do.
  • It's preferable to do all the complex joins and simple data manipulations with SQL.
  • It's preferable to aggregate JSONs in DB for responses with nested objects.
# src.posts.service
from typing import Mapping

from pydantic import UUID4
from sqlalchemy import desc, func, select, text
from sqlalchemy.sql.functions import coalesce

from src.database import databse, posts, profiles, post_review, products

async def get_posts(
    creator_id: UUID4, *, limit: int = 10, offset: int = 0
) -> list[Mapping]: 
    select_query = (
        select(
            (
                posts.c.id,
                posts.c.type,
                posts.c.slug,
                posts.c.title,
                func.json_build_object(
                   text("'id', profiles.id"),
                   text("'first_name', profiles.first_name"),
                   text("'last_name', profiles.last_name"),
                   text("'username', profiles.username"),
                ).label("creator"),
            )
        )
        .select_from(posts.join(profiles, posts.c.owner_id == profiles.c.id))
        .where(posts.c.owner_id == creator_id)
        .limit(limit)
        .offset(offset)
        .group_by(
            posts.c.id,
            posts.c.type,
            posts.c.slug,
            posts.c.title,
            profiles.c.id,
            profiles.c.first_name,
            profiles.c.last_name,
            profiles.c.username,
            profiles.c.avatar,
        )
        .order_by(
            desc(coalesce(posts.c.updated_at, posts.c.published_at, posts.c.created_at))
        )
    )
    
    return await database.fetch_all(select_query)

# src.posts.schemas
import orjson
from enum import Enum

from pydantic import BaseModel, UUID4, validator


class PostType(str, Enum):
    ARTICLE = "ARTICLE"
    COURSE = "COURSE"

   
class Creator(BaseModel):
    id: UUID4
    first_name: str
    last_name: str
    username: str


class Post(BaseModel):
    id: UUID4
    type: PostType
    slug: str
    title: str
    creator: Creator

    @validator("creator", pre=True)  # before default validation
    def parse_json(cls, creator: str | dict | Creator) -> dict | Creator:
       if isinstance(creator, str):  # i.e. json
          return orjson.loads(creator)

       return creator
    
# src.posts.router
from fastapi import APIRouter, Depends

router = APIRouter()


@router.get("/creators/{creator_id}/posts", response_model=list[Post])
async def get_creator_posts(creator: Mapping = Depends(valid_creator_id)):
   posts = await service.get_posts(creator["id"])

   return posts

If an aggregated data form DB is a simple JSON, then take a look at Pydantic's

Json
field type, which will load raw JSON first.

from pydantic import BaseModel, Json

class A(BaseModel):
    numbers: Json[list[int]]
    dicts: Json[dict[str, int]]

valid_a = A(numbers="[1, 2, 3]", dicts='{"key": 1000}')  # becomes A(numbers=[1,2,3], dicts={"key": 1000})
invalid_a = A(numbers='["a", "b", "c"]', dicts='{"key": "str instead of int"}')  # raises ValueError

20. Validate hosts, if users can send publicly available URLs

For example, we have a specific endpoint which:

  1. accepts media file from the user,
  2. generates unique url for this file,
  3. returns url to user,
    1. which they will use in other endpoints like
      PUT /profiles/me
      ,
      POST /posts
    2. these endpoints accept files only from whitelisted hosts
  4. uploads file to AWS with this name and matching URL.

URL ホストをホワイトリストに登録しないと、悪いユーザーが危険なリンクをアップロードする可能性があります。

from pydantic import AnyUrl, BaseModel

ALLOWED_MEDIA_URLS = {"mysite.com", "mysite.org"}

class CompanyMediaUrl(AnyUrl):
    @classmethod
    def validate_host(cls, parts: dict) -> tuple[str, str, str, bool]:
       """Extend pydantic's AnyUrl validation to whitelist URL hosts."""
        host, tld, host_type, rebuild = super().validate_host(parts)
        if host not in ALLOWED_MEDIA_URLS:
            raise ValueError(
                "Forbidden host url. Upload files only to internal services."
            )

        return host, tld, host_type, rebuild


class Profile(BaseModel):
    avatar_url: CompanyMediaUrl  # only whitelisted urls for avatar

21. スキーマが直接クライアントに面している場合、カスタム pydantic バリデータで ValueError を発生させます

ユーザーに詳細な応答を返します。

# src.profiles.schemas
from pydantic import BaseModel, validator

class ProfileCreate(BaseModel):
    username: str
    
    @validator("username")
    def validate_bad_words(cls, username: str):
        if username  == "me":
            raise ValueError("bad username, choose another")
        
        return username


# src.profiles.routes
from fastapi import APIRouter

router = APIRouter()


@router.post("/profiles")
async def get_creator_posts(profile_data: ProfileCreate):
   pass

応答例:

22. FastAPI が Response Pydantic Object を Dict に変換し、次に ResponseModel のインスタンスに変換し、次に Dict に変換し、JSON に変換することを忘れないでください

from fastapi import FastAPI
from pydantic import BaseModel, root_validator

app = FastAPI()


class ProfileResponse(BaseModel):
    @root_validator
    def debug_usage(cls, data: dict):
        print("created pydantic model")

        return data

    def dict(self, *args, **kwargs):
        print("called dict")
        return super().dict(*args, **kwargs)


@app.get("/", response_model=ProfileResponse)
async def root():
    return ProfileResponse()

ログ出力:

[INFO] [2022-08-28 12:00:00.000000] created pydantic model
[INFO] [2022-08-28 12:00:00.000010] called dict
[INFO] [2022-08-28 12:00:00.000020] created pydantic model
[INFO] [2022-08-28 12:00:00.000030] called dict

23. 同期 SDK を使用する必要がある場合は、スレッド プールで実行します。

SDK を使用して外部サービスと対話する必要があり、そうでない場合

async
は、外部ワーカー スレッドで HTTP 呼び出しを行います。

簡単な例として、スターレットのよく知られているものを使用できます

run_in_threadpool

from fastapi import FastAPI
from fastapi.concurrency import run_in_threadpool
from my_sync_library import SyncAPIClient 

app = FastAPI()


@app.get("/")
async def call_my_sync_library():
    my_data = await service.get_my_data()

    client = SyncAPIClient()
    await run_in_threadpool(client.make_request, data=my_data)

24.リンターを使用する(黒、アイソルト、オートフレーク)

リンターを使用すると、コードの書式設定を忘れて、ビジネス ロジックの記述に集中できます。

Black は妥協のないコード フォーマッタであり、開発中に行わなければならない非常に多くの小さな決定を排除します。他のリンターは、よりクリーンなコードを記述し、PEP8 に従うのに役立ちます。

pre-commit フックを使用することは一般的な良い方法ですが、スクリプトを使用するだけでも問題ありませんでした。

#!/bin/sh -e
set -x

autoflake --remove-all-unused-imports --recursive --remove-unused-variables --in-place src tests --exclude=__init__.py
isort src tests --profile black
black src tests

ボーナスセクション

何人かの非常に親切な人々が、読む価値のある自分の経験とベスト プラクティスを共有してくれました。プロジェクトの問題セクションでそれらをチェックしてください。

たとえば、lowercase00 は、アクセス許可と認証、クラスベースのサービスとビュー、タスク キュー、カスタム レスポンス シリアライザー、dynaconf を使用した構成などを使用したベスト プラクティスを詳細に説明しています。

FastAPI を使用した経験について何か共有したいことがあれば、それが良いか悪いかにかかわらず、新しいイシューを作成してください。それを読むのは私たちの喜びです。