12
7

More than 1 year has passed since last update.

FastAPIとDjango ORMを組み合わせる

Last updated at Posted at 2020-08-21

動機

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

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

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

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

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

フォルダ構造

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

django-admin startproject mysite
django-admin startapp poll

ファイルを生成した後、views.pymodels.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にもある。

12
7
2

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
12
7