当前位置:   article > 正文

像Django一样开发FastAPI之: AppBoot 入门指南

像Django一样开发FastAPI之: AppBoot 入门指南

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
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12

新建APP polls

python manage.py startapp polls
  • 1

创建数据库 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]
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7

创建 Schema

polls/schema.py 中定义 QuestionSchema

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

class QuestionSchema(ModelSchema):
    class Meta:
        model = Question
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6

编写 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()
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29

配置 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'])
  • 1
  • 2
  • 3
  • 4
  • 5

测试 API

python manage.py runserver
  • 1

现在可以通过浏览器直接访问我们的 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 字段模糊查询
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7

polls/views.py 文件中将 QuerySchema 替换为 QuestionQuerySchema,然后在浏览器中刷新文档页面,你会看到question列表接口增加了两个新的查询参数。

query

尝试示例

访问 Examples 获取完整示例,如果你觉得这个项目对你有帮助或有趣,请点赞支持一下!你的支持是我们继续改进和发展的动力。感谢你的关注!!!

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/寸_铁/article/detail/966889
推荐阅读
相关标签
  

闽ICP备14008679号