第263期 fastAPI 教程这一篇就够

项目结构

项目结构有很多种，但最好的结构是一致、直观且没有意外的。

许多示例项目和教程按文件类型（如crud、routers、models）划分项目，这种方式对于微服务或范围较小的项目很有效。但是，这种方法并不适合我们这个包含许多领域和模块的单体应用。

我发现对于这类情况，更具可扩展性和可演进性的结构是受Netflix的Dispatch启发，并做了一些小修改。

fastapi-project
├── alembic/
├── src
│   ├── auth
│   │   ├── router.py
│   │   ├── schemas.py  # pydantic模型
│   │   ├── models.py  # 数据库模型
│   │   ├── dependencies.py
│   │   ├── config.py  # 本地配置
│   │   ├── constants.py
│   │   ├── exceptions.py
│   │   ├── service.py
│   │   └── utils.py
│   ├── aws
│   │   ├── client.py  # 用于外部服务通信的客户端模型
│   │   ├── schemas.py
│   │   ├── config.py
│   │   ├── constants.py
│   │   ├── exceptions.py
│   │   └── utils.py
│   └── posts
│   │   ├── router.py
│   │   ├── schemas.py
│   │   ├── models.py
│   │   ├── dependencies.py
│   │   ├── constants.py
│   │   ├── exceptions.py
│   │   ├── service.py
│   │   └── utils.py
│   ├── config.py  # 全局配置
│   ├── models.py  # 全局模型
│   ├── exceptions.py  # 全局异常
│   ├── pagination.py  # 全局模块，如分页
│   ├── database.py  # 数据库连接相关内容
│   └── main.py
├── tests/
│   ├── auth
│   ├── aws
│   └── posts
├── templates/
│   └── index.html
├── requirements
│   ├── base.txt
│   ├── dev.txt
│   └── prod.txt
├── .env
├── .gitignore
├── logging.ini
└── alembic.ini

将所有领域目录存储在src文件夹中
1. src/ - 应用的最高级别，包含通用模型、配置和常量等。
2. src/main.py - 项目的根文件，用于初始化FastAPI应用
每个包都有自己的路由、模式、模型等。
1. router.py - 每个模块的核心，包含所有端点
2. schemas.py - 用于pydantic模型
3. models.py - 用于数据库模型
4. service.py - 模块特定的业务逻辑
5. dependencies.py - 路由依赖项
6. constants.py - 模块特定的常量和错误代码
7. config.py - 例如环境变量
8. utils.py - 非业务逻辑函数，例如响应规范化、数据丰富等
9. exceptions.py - 模块特定的异常，例如PostNotFound、InvalidUserData
当包需要其他包的服务、依赖项或常量时，使用显式的模块名导入

from src.auth import constants as auth_constants
from src.notifications import service as notification_service
from src.posts.constants import ErrorCode as PostsErrorCode  # 以防每个包的constants模块中都有标准的ErrorCode

异步路由

FastAPI首先是一个异步框架。它设计用于处理异步I/O操作，这也是它如此快速的原因。

然而，FastAPI并不限制你只能使用async路由，开发者也可以使用同步路由。这可能会让初学者误以为它们是一样的，但实际上并非如此。

I/O密集型任务

在底层，FastAPI可以有效地处理异步和同步I/O操作。

FastAPI在线程池中运行同步路由，阻塞的I/O操作不会阻止事件循环执行任务。
如果路由定义为async，那么它会通过await正常调用，FastAPI相信你只会执行非阻塞的I/O操作。

需要注意的是，如果你违反了这种信任，在异步路由中执行阻塞操作，事件循环将无法在阻塞操作完成之前运行后续任务。

import asyncio
import time

from fastapi import APIRouter

router = APIRouter()

@router.get("/terrible-ping")
async def terrible_ping():
    time.sleep(10) # 10秒的I/O阻塞操作，整个进程都会被阻塞

    return {"pong": True}

@router.get("/good-ping")
def good_ping():
    time.sleep(10) # 10秒的I/O阻塞操作，但在单独的线程中运行整个`good_ping`路由

    return {"pong": True}

@router.get("/perfect-ping")
async def perfect_ping():
    await asyncio.sleep(10) # 非阻塞I/O操作

    return {"pong": True}

目录

项目结构

异步路由

I/O密集型任务