AppBoot：像 Django 一样使用 FastAPI

2024/10/23 23:28:49 来源：https://blog.csdn.net/li_yatao/article/details/140940372 浏览: 次关键词：AppBoot：像 Django 一样使用 FastAPI

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 modelsclass 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 Questionclass 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 QuestionSchemarouter = 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 routerroot_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, ContainsFieldclass QuestionQuerySchema(QuerySchema):ids: Optional[list[int]] = EqField(None, alias='pk', columns='id')  # 按 ID 列表查询 Questionquestion_text: Optional[str] = ContainsField(None)  # question_text 字段模糊查询