Metadata-Version: 2.3
Name: fastapirestframework-tortoise
Version: 0.0.2
Summary: DRF-inspired async CRUD framework for FastAPI and Tortoise ORM
Requires-Dist: fastapi==0.146.0
Requires-Dist: pydantic==2.14.1
Requires-Dist: tortoise-orm==1.1.9
Requires-Python: >=3.12
Description-Content-Type: text/markdown

# FastAPI REST Framework · Tortoise

`fastapirestframework-tortoise` 是基于 FastAPI、Tortoise ORM 和 Pydantic v2 的异步 REST API 开发库，提供 DRF 风格的 Serializer、ViewSet、Router，以及按模型自动生成的 Vue 3 管理后台。

- PyPI 发行名：`fastapirestframework-tortoise`，当前版本 **0.0.2**。
- Python 导入名：`fastapirestframework`。
- Python 要求：**3.12 或更高**。当前运行依赖固定为 FastAPI 0.146.0、Pydantic 2.14.1、Tortoise ORM 1.1.9。
- 后台资源随 wheel 提供，运行无需 Node、npm 或 CDN。

## 安装与启动

在一个新目录中执行（已安装 uv）：

```bash
uv init --python 3.12
uv add fastapirestframework-tortoise==0.0.2 uvicorn
```

将下一节完整代码复制保存为当前目录的 **app.py**，然后运行：

```bash
uv run uvicorn app:app --host 127.0.0.1 --port 8000
```

打开：

- [自动管理后台](http://127.0.0.1:8000/admin/)：先新增用户、标签，再新增文章。
- [Swagger API 文档](http://127.0.0.1:8000/docs)：点击 Authorize，输入 `admin-demo` 或 `user-demo`，不需要手动输入 Bearer 前缀。

按 Ctrl+C 停止。示例使用 SQLite 内存数据库，启动时自动建表，重启后数据清空。生产应用应自行管理数据库配置和迁移。

## 完整单文件 Demo

关系为：User 一对多 Post；每篇 Post 必须有一个作者，可以有零个或多个 Tag；Tag 可由多篇文章共享。User 仅为作者资料，与演示认证令牌分开。

这份代码无需导入仓库的 `examples`，可以独立保存为 `app.py` 使用。

```python
"""可独立复制的单文件入门 Demo：API、认证权限限流及自动后台。"""

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import ClassVar

from fastapi import Depends, FastAPI, Request
from fastapi.security import HTTPBearer
from tortoise import fields
from tortoise.contrib.fastapi import RegisterTortoise
from tortoise.models import Model
from tortoise.queryset import QuerySet
from tortoise.transactions import in_transaction

from fastapirestframework import configure
from fastapirestframework.admin import AdminRouter, FieldOptions, ModelAdmin
from fastapirestframework.authentication import BearerAuthentication, Identity
from fastapirestframework.decorators import action
from fastapirestframework.exceptions import AuthenticationFailed, Conflict, ValidationError
from fastapirestframework.permissions import IsAdminUser, IsAuthenticated
from fastapirestframework.routers import DefaultRouter
from fastapirestframework.serializers import (
    Field,
    IntegerField,
    ModelSerializer,
    NestedField,
    Serializer,
)
from fastapirestframework.throttling import ScopedRateThrottle, UserRateThrottle
from fastapirestframework.viewsets import ModelViewSet

DEMO_TOKENS: dict[str, Identity] = {
    "user-demo": Identity(user="demo-user", is_authenticated=True),
    "admin-demo": Identity(user="demo-admin", is_authenticated=True, is_admin=True),
}


class DemoAuthentication(BearerAuthentication):
    """固定令牌仅用于演示认证，User 是作者资料，不承担登录账户职责。"""

    async def authenticate_credentials(self, credential: str, request: Request) -> Identity:
        """未知令牌立即401，不把客户端自报用户当作已验证身份。"""
        identity: Identity | None = DEMO_TOKENS.get(credential)
        if identity is None:
            raise AuthenticationFailed("演示令牌无效。")
        return identity


class User(Model):
    """一个作者可以有多篇文章；已有文章的作者不能直接删除。"""

    id = fields.IntField(primary_key=True)
    username = fields.CharField(max_length=50, unique=True)


class Tag(Model):
    """标签可以独立存在，也可以被多篇文章共享。"""

    id = fields.IntField(primary_key=True)
    name = fields.CharField(max_length=50, unique=True)


class Post(Model):
    """文章必须有一个作者；中间表保存零个或多个不重复的标签关系。"""

    id = fields.IntField(primary_key=True)
    title = fields.CharField(max_length=200)
    content = fields.TextField()
    author: fields.ForeignKeyRelation[User] = fields.ForeignKeyField(
        "models.User", related_name="posts", null=False, on_delete=fields.OnDelete.RESTRICT
    )
    tags: fields.ManyToManyRelation[Tag] = fields.ManyToManyField(
        "models.Tag",
        related_name="posts",
        through="post_tags",
        on_delete=fields.OnDelete.CASCADE,
        unique=True,
    )


class UserSerializer(ModelSerializer):
    """作者白名单；用户名唯一性由数据库约束保证。"""

    class Meta:
        model: type[Model] = User
        fields: list[str] = ["id", "username"]
        read_only_fields: list[str] = ["id"]


class TagSerializer(ModelSerializer):
    """标签只暴露id和name，不循环嵌套关联文章。"""

    class Meta:
        model: type[Model] = Tag
        fields: list[str] = ["id", "name"]
        read_only_fields: list[str] = ["id"]


class PostSerializer(ModelSerializer):
    """写入已有作者/标签ID，读取展开对象；不支持嵌套创建关联记录。"""

    # 显式声明author_id，注册期无需等待ORM生成外键的底层标量字段。
    author_id = IntegerField(write_only=True, gt=0)
    tag_ids = Field(list[int], default=list, write_only=True)
    author = NestedField(UserSerializer)
    tags = NestedField(TagSerializer, many=True)

    class Meta:
        model: type[Model] = Post
        fields: list[str] = ["id", "title", "content", "author_id", "tag_ids", "author", "tags"]
        read_only_fields: list[str] = ["id"]

    def validate_tag_ids(self, value: object) -> list[int]:
        """拒绝非正整数和重复标签，避免无意重复绑定。"""
        if not isinstance(value, list):
            raise ValidationError({"tag_ids": ["必须是标签ID列表。"]})
        ids: list[int] = []
        for item in value:
            if not isinstance(item, int) or item <= 0:
                raise ValidationError({"tag_ids": ["标签ID必须是正整数。"]})
            ids.append(item)
        if len(set(ids)) != len(ids):
            raise ValidationError({"tag_ids": ["标签ID不能重复。"]})
        return ids

    async def create(self, data: dict[str, object]) -> Post:
        """创建文章和标签关联共享一个事务。"""
        return await self._write(None, data)

    async def update(self, instance: object, data: dict[str, object]) -> Post:
        """PATCH省略关联不修改；空列表清空；列表替换全部标签。"""
        if not isinstance(instance, Post):
            raise TypeError("PostSerializer只更新Post实例")
        return await self._write(instance, data)

    async def _write(self, instance: Post | None, data: dict[str, object]) -> Post:
        """先验证引用，再保存；所有查询、写入和中间表操作使用同一连接。"""
        values: dict[str, object] = dict(data)
        tag_ids: list[int] | None = (
            self.validate_tag_ids(values.pop("tag_ids")) if "tag_ids" in values else None
        )
        async with in_transaction() as connection:
            if (
                "author_id" in values
                and not await User.filter(id=values["author_id"]).using_db(connection).exists()
            ):
                raise ValidationError({"author_id": ["作者不存在。"]})
            tags: list[Tag] = []
            if tag_ids is not None:
                tags = await Tag.filter(id__in=tag_ids).using_db(connection).order_by("id")
                if len(tags) != len(tag_ids):
                    raise ValidationError({"tag_ids": ["存在未找到的标签。"]})
            post: Post
            if instance is None:
                post = Post(**values)
                await post.save(using_db=connection, force_create=True)
            else:
                # 使用事务中的新实例，不让回滚后旧对象残留未提交的内存字段值。
                post = await Post.filter(id=instance.pk).using_db(connection).get()
                if values:
                    post.update_from_dict(values)
                    await post.save(using_db=connection, update_fields=list(values))
            if tag_ids is not None:
                await post.tags.clear(using_db=connection)
                if tags:
                    await post.tags.add(*tags, using_db=connection)
            return post


class UserViewSet(ModelViewSet[User]):
    """作者CRUD；拒绝删除仍被文章引用的作者。"""

    serializer_class: ClassVar[type[Serializer]] = UserSerializer

    def get_queryset(self) -> QuerySet[User]:
        """请求内构造查询，不在导入时连接数据库。"""
        return User.all().order_by("id")

    async def perform_destroy(self, instance: User) -> None:
        """业务409说明原因，数据库RESTRICT同时保证作者引用不悬空。"""
        async with in_transaction() as connection:
            if await Post.filter(author_id=instance.pk).using_db(connection).exists():
                raise Conflict("用户仍有文章，请先转移作者或删除文章。")
            await instance.delete(using_db=connection)


class TagViewSet(ModelViewSet[Tag]):
    """标签CRUD；删除标签由数据库清理中间表，文章仍保留。"""

    serializer_class: ClassVar[type[Serializer]] = TagSerializer

    def get_queryset(self) -> QuerySet[Tag]:
        """仅返回标签，不自动循环展开文章。"""
        return Tag.all().order_by("id")


class PostViewSet(ModelViewSet[Post]):
    """文章CRUD复用框架，关联写入交给PostSerializer扩展点。"""

    serializer_class: ClassVar[type[Serializer]] = PostSerializer

    def get_queryset(self) -> QuerySet[Post]:
        """NestedField会批量预加载作者与标签，不在同步输出时查询。"""
        return Post.all().order_by("id")

    @action(
        detail=False,
        permission_classes=[IsAdminUser],
        throttle_classes=[ScopedRateThrottle],
        throttle_scope="statistics",
        response_model=dict[str, int],
        summary="管理员统计文章数量",
    )
    async def statistics(self) -> dict[str, int]:
        """普通用户403；管理员统计独立限流。"""
        return {"total": await self.get_queryset().count()}


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    """应用持有唯一数据库生命周期；内存库重启清空，仅演示时自动建表。"""
    async with RegisterTortoise(
        app=app,
        db_url="sqlite://:memory:",
        modules={"models": [__name__]},
        generate_schemas=True,
    ):
        yield


app: FastAPI = FastAPI(title="FRF User / Post / Tag Demo", lifespan=lifespan)
# 为便于连续操作关联流程，默认100/min；修改速率后重启即可体验更低限额。
configure(
    default_authentication_classes=[DemoAuthentication],
    default_permission_classes=[IsAuthenticated],
    default_throttle_classes=[UserRateThrottle],
    default_throttle_rates={"user": "100/min", "statistics": "2/min"},
)
router: DefaultRouter = DefaultRouter()
# 仅给Swagger提供令牌输入框，凭证仍由DemoAuthentication验证。
security: HTTPBearer = HTTPBearer(auto_error=False, scheme_name="DemoBearer")
router.register("users", UserViewSet, basename="user", dependencies=[Depends(security)])
router.register("tags", TagViewSet, basename="tag", dependencies=[Depends(security)])
router.register("posts", PostViewSet, basename="post", dependencies=[Depends(security)])
app.include_router(router)


# 后台复用同一模型和数据库生命周期；其表单使用author/tags关系名称。
admin: AdminRouter = AdminRouter(prefix="/admin", title="FRF 入门后台")
admin.register(
    User,
    slug="users",
    admin=ModelAdmin(
        label="用户",
        icon="UserFilled",
        fields=("id", "username"),
        list_display=("id", "username"),
        search_fields=("username",),
        label_field="username",
    ),
)
admin.register(
    Tag,
    slug="tags",
    admin=ModelAdmin(
        label="标签",
        icon="CollectionTag",
        fields=("id", "name"),
        list_display=("id", "name"),
        search_fields=("name",),
        label_field="name",
    ),
)
admin.register(
    Post,
    slug="posts",
    admin=ModelAdmin(
        label="文章",
        icon="Document",
        fields=("id", "title", "content", "author", "tags"),
        list_display=("id", "title", "author", "tags"),
        list_editable=("title", "author"),
        search_fields=("title", "content"),
        list_filter=("author", "tags"),
        ordering=("-id", "title"),
        label_field="title",
        field_options={
            "title": FieldOptions(label="标题"),
            "content": FieldOptions(label="正文", widget="textarea"),
            "author": FieldOptions(label="作者"),
            "tags": FieldOptions(label="标签"),
        },
    ),
)
app.include_router(admin)

```

## 调用 API

下列命令适用于新启动、尚无数据的 Demo。先创建作者和标签，再创建文章。示例ID均为1；已有数据时请使用响应中的实际ID。

```bash
curl -X POST http://127.0.0.1:8000/users/ \
  -H 'Authorization: Bearer admin-demo' \
  -H 'Content-Type: application/json' \
  -d '{"username":"alice"}'

curl -X POST http://127.0.0.1:8000/tags/ \
  -H 'Authorization: Bearer admin-demo' \
  -H 'Content-Type: application/json' \
  -d '{"name":"Python"}'

curl -X POST http://127.0.0.1:8000/posts/ \
  -H 'Authorization: Bearer admin-demo' \
  -H 'Content-Type: application/json' \
  -d '{"title":"第一篇文章","content":"Hello FRF","author_id":1,"tag_ids":[1]}'

curl http://127.0.0.1:8000/posts/ \
  -H 'Authorization: Bearer admin-demo'

curl -X PATCH http://127.0.0.1:8000/posts/1/ \
  -H 'Authorization: Bearer admin-demo' \
  -H 'Content-Type: application/json' \
  -d '{"title":"修改标题","tag_ids":[]}'

curl http://127.0.0.1:8000/posts/statistics/ \
  -H 'Authorization: Bearer admin-demo'
```

创建文章返回201，响应展开关联对象，例如：

```json
{
  "id": 1,
  "title": "第一篇文章",
  "content": "Hello FRF",
  "author": {"id": 1, "username": "alice"},
  "tags": [{"id": 1, "name": "Python"}]
}
```

### CRUD 路由

将表格中的 `posts` 替换为 `users` 或 `tags` 可操作对应资源；三者均由 ModelViewSet 自动生成。

| 方法 | 路径 | 行为 |
| --- | --- | --- |
| GET | `/posts/` | 查询列表；当前Demo未启用普通API分页，返回数组 |
| POST | `/posts/` | 创建文章，返回201 |
| GET | `/posts/{pk}/` | 查看详情 |
| PUT | `/posts/{pk}/` | 完整更新 |
| PATCH | `/posts/{pk}/` | 局部更新 |
| DELETE | `/posts/{pk}/` | 删除，返回204且无响应体 |
| GET | `/posts/statistics/` | 集合Action，返回文章总数，仅管理员可用 |

PATCH 省略字段表示保持原值；`tag_ids: []` 清空标签，列表替换全部标签，`null` 不允许。作者必须存在，重复或无效标签返回400；作者仍有文章时删除返回409。

### 认证、权限和限流

- 不带令牌访问普通API返回401；`user-demo` 与 `admin-demo` 都可以操作普通CRUD。Demo未做作者隔离，二者共享数据。
- 统计Action仅允许 `admin-demo`；`user-demo` 调用返回403。`detail=False` 表示集合行为；`detail=True` 的自定义行为会绑定单条记录并先执行对象权限。
- 全局 `configure(...)` 必须在创建 Router 前调用。有效配置优先级为 Action → ViewSet → 全局默认；显式空策略列表可关闭对应策略。
- 普通API按身份100次/分钟；统计接口独立2次/分钟，超限返回429及可用的 Retry-After。内置计数仅在同一进程内共享，不是分布式限流。
- 固定令牌仅用于演示，生产中应继承认证器验证真实凭证并返回 Identity；框架不提供登录账户系统、JWT签发或撤销。

### Serializer 与关联保存

ModelSerializer 的 `Meta.model` 绑定 ORM 模型，`Meta.fields` 显式限制公开字段；只读字段拒绝写入，只写字段不出现在响应中。ModelViewSet 绑定 Serializer 并实现 `get_queryset()`，DefaultRouter 注册后提供标准CRUD。查询在请求内创建，应用自己管理 ORM 生命周期。

Demo 的 `author_id`、`tag_ids` 是输入字段，`author`、`tags` 通过 NestedField 输出展开对象。关联写入由 PostSerializer 的 create/update 扩展实现，引用检查、文章保存及中间表写入共用同一事务。NestedField 本身仅用于输出，不能据此认为框架支持任意嵌套写入。

需要增加业务校验时，可参考代码中的 `validate_tag_ids`；多次写入业务通过 `in_transaction()` 显式控制事务。同步序列化阶段不应临时发起异步数据库查询。

## 自动管理后台 frfadmin

`AdminRouter.register(Model, slug=..., admin=ModelAdmin(...))` 根据注册模型生成界面，无需为每个模型编写 Vue 页面。已有应用只需添加 Demo 末尾的后台注册部分，复用原有模型与数据库生命周期。

| 常用配置 | 用途 |
| --- | --- |
| `label`、`icon` | 导航名称与Element Plus图标名，未配图标时使用Grid |
| `fields`、`readonly_fields` | 表单及详情公开字段、只读字段 |
| `list_display` | 列表列及顺序 |
| `list_editable` | 双击可编辑的列表字段，使用列表编辑栏保存或取消 |
| `search_fields`、`list_filter` | 搜索字段、允许精确筛选的字段 |
| `ordering` | 默认排序与允许切换的字段 |
| `label_field` | 关联候选项的可读名称 |
| `field_options` | 中文标签、帮助文字、占位文字、textarea等控件 |
| `page_size` | 后台每页条数，默认20 |

后台支持搜索、筛选、排序、关联远程选择及快捷保存；分页选项为10、20、100、200、500、1000。导航图标采用 UserFilled / Document / CollectionTag。操作图标悬停显示名称，详情完整内容不重复提示，列表文字实际溢出才显示全文提示。

后台还提供 ModelAdmin 类继承与业务钩子、display 计算列、action 批量行为、Fieldset 动态分组、FK Inline、区间/空值/自定义筛选和浏览器本地保存查询等扩展点。本入门Demo仅启用代码中明确声明的配置，未配置全部扩展。

后台输入关系名为 `author` 和 `tags`，普通Demo API输入为 `author_id` 和 `tag_ids`，两套输入契约不同。后台使用ModelAdmin钩子，不自动调用普通PostSerializer的自定义钩子；新增业务约束需要在对应入口实施。

当前后台**没有内置登录**，独立于普通API的认证权限限流；应由部署应用决定访问保护。关联快捷保存为独立事务，取消主表单不撤销已快捷保存的关联记录。

## 能力与验证范围

框架还提供分页、白名单过滤排序、对象权限、计算字段等能力；本Demo未启用普通API分页或作者数据隔离，不能将后台配置直接当作普通API配置。

新入门Demo以SQLite内存库验收；历史框架曾验证MySQL 8.3.0，不代表本轮所有后台组合已验证。PostgreSQL及其他数据库/依赖组合未验证。框架不接管生产迁移，不提供账户系统或分布式限流。

此版本仅发布通用wheel：包含Python框架、预构建后台资源及本说明的包元数据。Demo完整源码已嵌入本页，可复制运行；无需寻找安装包内的examples目录。源码发行包的构建链问题仍待排查，本次不上传未经验证的sdist。
