AppBoot：像 Django 一样使用 FastAPI

news2024/11/13 9:10:55

App Boot

开发 AppBoot 的背景是我一直没能寻找到满意的 FastAPI 项目模板。相比之下，Django 的项目结构和开发方式一直深得我心，因此我决定创建一个类似 Django 的 FastAPI 项目模板。

AppBoot 完全采用异步模式，内置 SQLAlchemy 2.0，开箱即用，旨在提供熟悉且高效的开发体验。

技术栈

Python 3.9+
FastAPI
SQLAlchemy 2.0+
Pydantic 支持v1和v2
Uvicorn

快速开始

启动新项目

# 创建项目目录
mkdir mysite
cd mysite
# 创建虚拟环境以在本地隔离包依赖
python3 -m venv env
source env/bin/activate  # Windows 使用 `env\\Scripts\\activate`
# 安装 appboot 和 aiosqlite 到虚拟环境中
pip install appboot aiosqlite
# 使用单个应用程序设置新项目
appboot startproject mysite .  # 注意尾随的 '.' 字符
# 启动服务器，应用运行在 http://127.0.0.1:8000
python manage.py runserver

新建APP polls

python manage.py startapp polls

创建数据库 Model

在 polls/models.py 中定义 Question 模型。

from datetime import datetime
from sqlalchemy.orm import Mapped
from appboot import models

class Question(models.Model):
    question_text: Mapped[str]
    pub_date: Mapped[datetime]

创建 Schema

在 polls/schema.py 中定义 QuestionSchema。

from appboot.schema import ModelSchema
from polls.models import Question

class QuestionSchema(ModelSchema):
    class Meta:
        model = Question

编写 CRUD API

在 polls/views.py 中编写 CRUD API。

from fastapi import APIRouter, Depends
from appboot.db import create_tables
from appboot.params import QuerySchema, QueryDepends, PaginationResult
from polls.models import Question
from polls.schema import QuestionSchema

router = APIRouter(dependencies=[Depends(create_tables)])

@router.post('/questions/', response_model=QuestionSchema)
async def create_question(question: QuestionSchema):
    return await question.create()

@router.get('/questions/', response_model=PaginationResult[QuestionSchema])
async def query_questions(query: QuerySchema = QueryDepends()):
    return await query.query_result(Question.objects.clone())

@router.get('/questions/{question_id}', response_model=QuestionSchema)
async def get_question(question_id: int):
    return await Question.objects.get(question_id)

@router.put('/questions/{question_id}', response_model=QuestionSchema)
async def update_question(question_id: int, question: QuestionSchema):
    instance = await Question.objects.get(question_id)
    return await question.update(instance)

@router.delete('/questions/{question_id}', response_model=QuestionSchema)
async def delete_question(question_id: int):
    instance = await Question.objects.get(question_id)
    return await instance.delete()

配置 API 路由规则

在 mysite/urls.py 中配置 API 路由。

from fastapi import APIRouter
from polls.views import router

root_router = APIRouter()
root_router.include_router(router, prefix='/polls', tags=['polls'])

测试 API

python manage.py runserver

现在可以通过浏览器直接访问我们的 API 文档，URL 为 http://127.0.0.1:8000/docs/。
docs

创建复杂查询的 QuerySchema

在 polls/schema.py 中创建 QuestionQuerySchema 以进行复杂查询。

from typing import Optional
from appboot.params import QuerySchema
from appboot.filters import EqField, ContainsField

class QuestionQuerySchema(QuerySchema):
    ids: Optional[list[int]] = EqField(None, alias='pk', columns='id')  # 按 ID 列表查询 Question
    question_text: Optional[str] = ContainsField(None)  # question_text 字段模糊查询