FastAPIとDjango ORMを組み合わせる #Python

動機

最近 FastAPI が破竹の勢いで伸びているらしい。

Django から FastAPI に浮気をしたいけど、やはり Django やそのユーザーシステムを引き続き利用したい。欲張りに見えるが、実はそういういい都合もある。今回はどうやって Django ORM と FastAPI を組み合わせるかについて説明する。

当然デメリットもある。パーフォーマンス上は、Django ORMは一部の非同期しか対応していないので、時に性能面の影響が出る。

パーフォーマンス上げたいなら、別途 orm、gino、sqlalchemy 1.4+ などのモデルを作成するがよい。

Django 4.1より, Django ORMで非同期クエリが実行できる。ここで注意したいのは、非同期トランザクションは対応されておらず、そしてPostgreSQLのドライバはまだpsycopg2であること。

フォルダ構造

まず、フォルダ構成について話す。フォルダの内容は Django ドキュメントのチュートリアルに基づいて作成していい。詳細はここ。

django-admin startproject mysite
django-admin startapp poll

ファイルを生成した後、views.pyとmodels.pyを削除して、下記のように FastAPI 用のフォルダを用意する。

$ tree -L 3 -I '__pycache__|venv' -P '*.py'
.
├── manage.py
├── mysite
│   ├── __init__.py
│   ├── asgi.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
└── polls
    ├── __init__.py
    ├── adapters
    │   └── __init__.py
    ├── admin.py
    ├── apps.py
    ├── migrations
    │   └── __init__.py
    ├── models
    │   └── __init__.py
    ├── routers
    │   └── __init__.py
    ├── schemas
    │   └── __init__.py
    └── tests.py

7 directories, 15 files

各フォルダの使い分け：

modelsフォルダ：Django ORM
routersフォルダ：FastAPI routers
schemasフォルダ：FastAPI の Pydantic バリデータ
adaptersフォルダ：Django ORM を取得するアダプター

ORM を使う FastAPI のウェブアプリにとって、ORM と Pydantic モデルの両方が存在する。いかに ORM を Pydantic モデルに変換するかというと、Pydantic の ORM モードを利用すればよい。

データを用意

Django ドキュメントを参考にして、データを入れてみよう。

>>> from polls.models import Choice, Question
>>> from django.utils import timezone
>>> q = Question(question_text="What's new?", pub_date=timezone.now())
>>> q.save()

FastAPI 導入

簡略化するため、一部のimportを省略する。

`schemas`

from django.db import models
from pydantic import BaseModel as _BaseModel

class BaseModel(_BaseModel):
    @classmethod
    def from_orms(cls, instances: List[models.Model]):
        return [cls.from_orm(inst) for inst in instances]


class FastQuestion(BaseModel):
    question_text: str
    pub_date: datetime

    class Config:
        orm_mode = True


class FastQuestions(BaseModel):
    items: List[FastQuestion]

    @classmethod
    def from_qs(cls, qs):
        return cls(items=FastQuestion.from_orms(qs))


class FastChoice(BaseModel):
    question: FastQuestion
    choice_text: str

    class Config:
        orm_mode = True


class FastChoices(BaseModel):
    items: List[FastChoice]

    @classmethod
    def from_qs(cls, qs):
        return cls(items=FastChoice.from_orms(qs))

`adapters`

Django 4.1より、非同期クエリが使えるようになった。

ModelT = TypeVar("ModelT", bound=models.Model)


async def retrieve_object(model_class: Type[ModelT], id: int) -> ModelT:
    instance = await model_class.objects.filter(pk=id).afirst()
    if not instance:
        raise HTTPException(status_code=404, detail="Object not found.")
    return instance


async def retrieve_question(q_id: int = Path(..., description="get question from db")):
    return await retrieve_object(Question, q_id)


async def retrieve_choice(c_id: int = Path(..., description="get choice from db")):
    return await retrieve_object(Choice, c_id)


async def retrieve_questions():
    return [q async for q in Question.objects.all()]


async def retrieve_choices():
    return [c async for c in Choice.objects.all()]

`routers`

`routers/init.py`

from .choices import router as choices_router
from .questions import router as questions_router

__all__ = ("register_routers",)


def register_routers(app: FastAPI):
    app.include_router(questions_router)
    app.include_router(choices_router)

`routers/choices.py`

router = APIRouter(prefix="/choice", tags=["choices"])


@router.get("/", response_model=FastChoices)
def get_choices(
    choices: List[Choice] = Depends(adapters.retrieve_choices),
) -> FastChoices:
    return FastChoices.from_qs(choices)


@router.get("/{c_id}", response_model=FastChoice)
def get_choice(choice: Choice = Depends(adapters.retrieve_choice)) -> FastChoice:
    return FastChoice.from_orm(choice)

`routers/questions.py`

router = APIRouter(prefix="/question", tags=["questions"])


@router.get("/", response_model=FastQuestions)
def get_questions(
    questions: List[Question] = Depends(adapters.retrieve_questions),
) -> FastQuestions:
    return FastQuestions.from_qs(questions)


@router.get("/{q_id}", response_model=FastQuestion)
def get_question(
    question: Question = Depends(adapters.retrieve_question),
) -> FastQuestion:
    return FastQuestion.from_orm(question)

`asgi.py`

mysite/asgi.pyに、FastAPI の App の起動も追加。

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

fastapp = FastAPI()


def init(app: FastAPI):
    from polls.routers import register_routers

    register_routers(app)

    if settings.MOUNT_DJANGO_APP:
        app.mount("/django", application)  # type:ignore
        app.mount("/static", StaticFiles(directory="staticfiles"), name="static")


init(fastapp)

立ち上げ

まず、uvicorn用のスタティックファイルを作成（whitenoiseも必要）：

python manage.py collectstatic --noinput

FastAPI はuvicorn mysite.asgi:fastapp --reloadで、Django はuvicorn mysite.asgi:application --port 8001 --reloadで起動。

FastAPI の Doc はhttp://127.0.0.1:8000/docs/に、Django の admin 画面はhttp://127.0.0.1:8001/admin/にアクセス。

もしASGIアプリを一つだけにするなら、DjangoアプリはFastAPIにマウント:

# in mysite/settings.py

MOUNT_DJANGO_APP = True

Django管理画面はhttp://localhost:8000/django/adminにて。

まとめ

FastAPI と Django ORM の組み合わせは意外と簡単で、上手にインテグレーションの部分を分けると、すっきりとしたフォルダ構造もできるのである。

上のコードは筆者のGithubにもある。

FastAPIとDjango ORMを組み合わせる

動機

フォルダ構造

データを用意

FastAPI 導入

schemas

adapters

routers

routers/__init__.py

routers/choices.py

routers/questions.py

asgi.py