将FastAPI与Django ORM结合在一起

2 年 ago
清, 宇
3 minutes

動機 – Motivation

最近听说FastAPI正在迅猛发展。

我想要从Django转投至FastAPI，但我仍希望继续使用Django及其用户系统。尽管看起来有点贪心，但实际上也有这样的好处。本次将介绍如何结合使用Django ORM和FastAPI。

当然也有缺点。在性能方面，Django ORM仅支持部分异步操作，因此有时会对性能造成影响。
如果想提高性能，可以另外创建模型，比如orm、gino、sqlalchemy 1.4+等。
从Django 4.1开始，可以在Django ORM中执行异步查询。需要注意的是，异步事务尚未支持，并且仍然使用psycopg2作为PostgreSQL的驱动程序。

文件夹结构

首先，让我们来讨论文件夹的结构。可以根据 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 Web应用来说，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。

架构

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

适配器

从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()]

路由器

路由器/__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)

路由器/选择.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)

路由器/问题.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

在我的网站/asgi.py文件中，添加了FastAPI应用程序的启动。

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库启动方式为`uvicorn mysite.asgi:fastapp –reload`，而Django的启动方式为`uvicorn mysite.asgi:application –port 8001 –reload`。

FastAPI的文档访问地址为http://127.0.0.1:8000/docs/，Django的管理员界面可通过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上找到。

bannerAds