fastapi-rest-toolkit 0.1.1

类DRF风格的FastAPI工具包

FastAPI REST Toolkit

FastAPI REST Toolkit 是一个类 Django REST Framework 风格的 FastAPI 工具包。

它围绕 SQLAlchemy async model、sqlalchemy-crud-plus 和 FastAPI router，提供 ViewSet、自动路由、认证、权限、过滤、排序、分页、限流和自定义 action 等常用 REST API 能力。

特性

• ViewSet：提供 list、retrieve、create、update、destroy 标准 CRUD action
• DefaultRouter：按 ViewSet 自动注册 REST 路由
• Schema 自动生成：可从 SQLAlchemy model 生成 Pydantic schema
• 认证：提供 BaseAuthentication 和 BearerAuthentication
• 权限：内置 AllowAny、IsAuthenticated、IsAdmin
• 过滤：支持普通 query 参数过滤、search 搜索和 ordering 排序
• 分页：内置 limit/offset 分页
• 关系加载：透传 load_strategies、join_conditions 给 sqlalchemy-crud-plus
• 限流：支持内存限流和 Redis 异步限流
• 自定义 action：支持类似 DRF 的 @action

安装

pip install fastapi-rest-toolkit

Redis 限流需要安装可选依赖：

pip install "fastapi-rest-toolkit[redis]"

本项目当前要求 Python >=3.12。

本地运行 demo

在项目根目录运行：

uv run demo/main.py

启动后访问：

http://127.0.0.1:8000/docs

demo 包含两个资源：

• users：演示 Bearer JWT 认证、权限、Redis 限流、搜索、排序、关系预加载和自定义 action
• posts：演示匿名访问、匿名限流、搜索、排序和外键字段

demo 的主要路由：

POST   /auth/dev-token?user_id=1
GET    /api/users
POST   /api/users
GET    /api/users/{pk}
PUT    /api/users/{pk}
DELETE /api/users/{pk}
GET    /api/users/hello
GET    /api/posts
POST   /api/posts
GET    /api/posts/{pk}
PUT    /api/posts/{pk}
DELETE /api/posts/{pk}

快速开始

下面是最小结构，和 demo 的写法保持一致。

from fastapi import FastAPI
from sqlalchemy_crud_plus import CRUDPlus

from fastapi_rest_toolkit import AllowAny, CRUDService, DefaultRouter, ViewSet
from app.db.session import get_session
from app.models.post import Post


class PostViewSet(ViewSet):
    model = Post
    permission_classes = (AllowAny,)
    search_fields = ("title", "content")
    ordering_fields = ("id", "title", "created_at")

    def __init__(self):
        post_crud = CRUDPlus(Post)
        self.service = CRUDService(crud=post_crud, model=Post)


app = FastAPI()
router = DefaultRouter()

router.register(
    "posts",
    PostViewSet,
    get_session=get_session,
    tags=["posts"],
)

app.include_router(router.router, prefix="/api")

只要配置了 model，ViewSet.init_schema() 会自动生成：

• read_schema：包含所有表字段
• create_schema：排除自增主键和 server_default 字段
• update_schema：所有字段都是可选字段

也可以显式配置自己的 read_schema、create_schema、update_schema。

Model 示例

demo 中的 Post 和 User 是普通 SQLAlchemy async model：

from datetime import datetime

from sqlalchemy import DateTime, ForeignKey, String, func
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.models.base import Base


class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str] = mapped_column(String(1000))
    author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

    author: Mapped["User"] = relationship(back_populates="posts")

如果只需要返回用户 id，author_id 已经是普通字段，不需要配置关系加载。

如果需要查询 posts 时顺便加载用户对象：

class PostViewSet(ViewSet):
    model = Post
    load_strategies = ("author",)

注意：自动生成的 schema 只包含表字段，不会自动序列化 relationship。要在响应中返回 author 的详细信息，需要自定义 read_schema 或重写 serialize()。

Router

DefaultRouter.register() 会实例化 ViewSet、初始化 schema、校验配置并注册路由。

router.register(
    prefix="users",
    viewset_cls=UserViewSet,
    get_session=get_session,
    tags=["users"],
    pk_type=int,
)

默认支持的 action：

Action	Method	Path
list	GET	/api/users
create	POST	/api/users
retrieve	GET	/api/users/{pk}
update	PUT	/api/users/{pk}
destroy	DELETE	/api/users/{pk}

可以通过 allowed_actions 控制启用的 action：

class ReadOnlyPostViewSet(ViewSet):
    model = Post
    allowed_actions = ("list", "retrieve")

ViewSet 配置

常用配置项：

class ExampleViewSet(ViewSet):
    model = None
    service = None

    read_schema = None
    create_schema = None
    update_schema = None

    authentication_classes = ()
    permission_classes = ()
    throttle_classes = ()

    filter_backends = (
        CRUDPlusFilterBackend(),
        SearchFilterBackend(),
        OrderingFilterBackend(),
    )
    search_fields = ()
    ordering_fields = ()

    pagination = LimitOffsetPagination()
    load_strategies = None
    join_conditions = None
    allowed_actions = ("list", "retrieve", "create", "update", "destroy")

service 必须提供标准 CRUD 方法。demo 使用 CRUDService(CRUDPlus(Model), Model)。

过滤、搜索和排序

默认启用三个 filter backend：

• CRUDPlusFilterBackend：把普通 query 参数透传给 sqlalchemy-crud-plus
• SearchFilterBackend：读取 search 参数，对 search_fields 做 LIKE
• OrderingFilterBackend：读取 ordering 参数，只允许 ordering_fields 中的字段

示例：

class PostViewSet(ViewSet):
    model = Post
    search_fields = ("title", "content")
    ordering_fields = ("id", "title", "created_at")

请求示例：

GET /api/posts?title__like=%FastAPI%
GET /api/posts?search=FastAPI
GET /api/posts?ordering=-created_at
GET /api/posts?search=FastAPI&ordering=title

分页

默认分页响应：

{
  "count": 1,
  "next": null,
  "previous": null,
  "results": []
}

请求参数：

GET /api/posts?limit=20&offset=0

自定义分页：

from fastapi_rest_toolkit import LimitOffsetPagination


class SmallPagination(LimitOffsetPagination):
    default_limit = 10
    max_limit = 50


class PostViewSet(ViewSet):
    pagination = SmallPagination()

认证和权限

认证类需要继承 BaseAuthentication 或 BearerAuthentication，并返回 (user, auth)。

demo 中的 UserAuthentication 会：

1. 从 Authorization: Bearer <token> 读取 token
2. 解码 JWT
3. 用当前 async session 查询用户
4. 把用户挂到 request.user

简化示例：

from typing import Any

from fastapi import HTTPException, status
from fastapi_rest_toolkit.authentication import BearerAuthentication
from fastapi_rest_toolkit.contextvar import session_var
from fastapi_rest_toolkit.request import FRFRequest


class UserAuthentication(BearerAuthentication):
    async def authenticate(self, request: FRFRequest) -> tuple[Any, Any]:
        session = session_var.get()
        token = self.get_token(request)
        if session is None or not token:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid authentication",
            )

        user = await load_user_from_token(session, token)
        return user, token

ViewSet 中使用：

class UserViewSet(ViewSet):
    authentication_classes = (UserAuthentication,)
    permission_classes = (IsAuthenticated,)

内置权限：

• AllowAny：允许所有请求
• IsAuthenticated：要求 request.user 不为空
• IsAdmin：要求 request.user.is_admin 为真

多个 permission 会依次检查，全部通过才允许继续执行。

限流

内置限流类：

• SimpleRateThrottle：内存限流
• AnonRateThrottle：只限制匿名请求
• AsyncRedisSimpleRateThrottle：Redis 异步限流，适合多进程或分布式部署

demo 的 posts 使用匿名限流：

from fastapi_rest_toolkit.throttle import AnonRateThrottle


AnonRateThrottle.THROTTLE_RATES.update({
    "anon": "10/minute",
})


class PostViewSet(ViewSet):
    throttle_classes = (AnonRateThrottle(),)

demo 的 users 使用 Redis 限流：

from fastapi_rest_toolkit.throttle import AsyncRedisSimpleRateThrottle
from app.db.redis import redis_client


class UserViewSet(ViewSet):
    throttle_classes = (AsyncRedisSimpleRateThrottle(redis=redis_client),)

限流速率格式：

100/day
10/hour
5/minute
1/second

关系加载

load_strategies 和 join_conditions 会透传给 sqlalchemy-crud-plus。

常见写法：

class UserViewSet(ViewSet):
    load_strategies = ("posts",)

更明确的写法：

class UserViewSet(ViewSet):
    load_strategies = {"posts": "selectinload"}

如果需要 join：

class UserViewSet(ViewSet):
    join_conditions = ["posts"]

或指定 join 类型：

class UserViewSet(ViewSet):
    join_conditions = {"posts": "left"}

优先使用 load_strategies 解决预加载问题。只有在需要基于关联表 join 查询时，再配置 join_conditions。

自定义 action

使用 @action 可以注册额外路由。

from fastapi_rest_toolkit import action


class UserViewSet(ViewSet):
    @action(methods=["get"], detail=False)
    async def hello(self, request, session):
        return {"message": "Hello, World!"}

上面的 action 会注册为：

GET /api/users/hello

detail action 会带上 pk：

class UserViewSet(ViewSet):
    @action(methods=["post"], detail=True, url_path="activate")
    async def activate(self, request, session, pk):
        return {"id": pk, "status": "activated"}

对应路由：

POST /api/users/{pk}/activate

异常处理

demo 注册了数据库完整性错误处理器，把唯一约束冲突转换成更友好的响应：

from sqlalchemy.exc import IntegrityError


def register_exception_handlers(app):
    app.add_exception_handler(IntegrityError, integrity_error_handler)

在 demo/main.py 中：

app = FastAPI(...)
register_exception_handlers(app)

Demo 目录

demo/
  main.py                 # FastAPI app 入口
  app/api/users.py        # UserViewSet
  app/api/posts.py        # PostViewSet
  app/auth/jwt.py         # JWT encode/decode
  app/deps/auth.py        # Bearer 认证实现
  app/db/session.py       # async SQLAlchemy session
  app/db/redis.py         # Redis client
  app/exceptions.py       # 异常处理
  app/models/             # SQLAlchemy models
  app/schemas/            # schema 生成示例

开发

同步依赖：

uv sync

运行 demo：

uv run demo/main.py

如果只想验证锁文件和依赖声明一致：

uv lock --locked

License

MIT License