fastapi-cbv 0.2.0
Django REST Framework style class-based views for FastAPI with Tortoise ORM support
License: MIT
Requires Python: >=3.8
Keywords: api, async, cbv, class-based-views, django-rest-framework, fastapi, rest, tortoise-orm

# FastAPI CBV (Class-Based Views)

[![Python Version](https://img.shields.io/badge/python-3.8%2B-blue)](https://www.python.org/)
[![FastAPI](https://img.shields.io/badge/FastAPI-0.68%2B-green)](https://fastapi.tiangolo.com/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

基于 FastAPI 的类视图第三方插件，完全参考 Django REST Framework 的设计理念，提供异步支持和 Tortoise ORM 集成。

## ✨ 特性

- 🚀 **完全异步**: 基于 FastAPI 和 async/await，支持高性能异步操作
- 🏗️ **Django REST Framework 风格**: 熟悉的 APIView、GenericAPIView、ViewSet 等概念
- 🔧 **Mixin 支持**: 可组合的 CreateModelMixin、ListModelMixin 等
- 🗄️ **Tortoise ORM 集成**: 深度集成 Tortoise ORM，自动序列化
- 📊 **自动分页**: 内置分页支持
- 🔍 **过滤和搜索**: 支持查询过滤和全文搜索
- 🔐 **认证系统**: 支持 JWT、Token、Basic、API Key 等多种认证方式
- 🛡️ **权限控制**: 类似 DRF 的权限类，支持 IsAuthenticated、IsAdminUser 等
- ⚠️ **异常处理**: 完善的异常类和处理器，返回统一的 JSON 错误响应
- 📝 **自动文档**: 完全兼容 FastAPI 的自动 API 文档生成
- 🎯 **类型安全**: 完整的类型注解支持，兼容 Pydantic V2

## 📦 安装

```bash
pip install fastapi-cbv
```

**依赖要求：**
- Python >= 3.8
- FastAPI >= 0.68.0
- Tortoise ORM >= 0.19.0 (可选，用于 ORM 集成)

## 🚀 快速开始

### 1. 基础 APIView

```python
from fastapi import FastAPI
from fastapi_cbv import APIView, cbv, CBVRouter

app = FastAPI()
router = CBVRouter()

@cbv(router)
class HelloView(APIView):
    async def get(self):
        return {"message": "Hello World"}
    
    async def post(self):
        data = await self.request.json()
        return {"received": data}

HelloView.add_api_route("/hello")
app.include_router(router)
```

### 2. 模型 CRUD 操作

```python
from tortoise.models import Model
from tortoise import fields
from fastapi_cbv import (
    ListCreateAPIView, 
    RetrieveUpdateDestroyAPIView,
    CBVRouter,
    create_tortoise_serializer
)

# 定义模型
class User(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=100)
    email = fields.CharField(max_length=100, unique=True)
    created_at = fields.DatetimeField(auto_now_add=True)

# 自动生成序列化器
UserSerializer = create_tortoise_serializer(User)
UserCreateSerializer = create_tortoise_serializer(
    User, name="UserCreate", exclude=["id", "created_at"]
)

router = CBVRouter()

# 列表和创建视图
class UserListView(ListCreateAPIView):
    serializer_class = UserSerializer
    
    def get_queryset(self):
        return User.all()
    
    def get_serializer_class(self):
        if self.request.method == "POST":
            return UserCreateSerializer
        return UserSerializer

# 详情、更新和删除视图
class UserDetailView(RetrieveUpdateDestroyAPIView):
    serializer_class = UserSerializer
    
    def get_queryset(self):
        return User.all()

# 注册路由
router.add_cbv_route("/users", UserListView)
router.add_cbv_route("/users/{id}", UserDetailView)
```

### 3. ViewSet 用法

```python
from fastapi_cbv import (
    ModelViewSet, 
    viewset_routes,
    TortoiseFilterBackend,
    TortoisePagination,
    action
)

class UserViewSet(ModelViewSet):
    serializer_class = UserSerializer
    filter_backends = [TortoiseFilterBackend]
    pagination_class = TortoisePagination
    search_fields = ['name', 'email']
    ordering_fields = ['id', 'name', 'created_at']
    ordering = ['-created_at']
    
    def get_queryset(self):
        return User.all()
    
    # 自定义 action
    @action(detail=False, methods=["get"])
    async def active(self, **kwargs):
        """获取所有激活的用户"""
        users = await User.filter(is_active=True).all()
        return [UserSerializer.model_validate(u) for u in users]
    
    @action(detail=True, methods=["post"])
    async def deactivate(self, id: int, **kwargs):
        """停用指定用户"""
        user = await User.get(id=id)
        user.is_active = False
        await user.save()
        return {"message": "User deactivated"}

# 自动生成所有 CRUD 路由
viewset_routes(router, UserViewSet, prefix="/users")
```

这将自动创建以下路由：
- `GET /users/` - 列表
- `POST /users/` - 创建
- `GET /users/{id}/` - 详情
- `PUT /users/{id}/` - 更新
- `PATCH /users/{id}/` - 部分更新
- `DELETE /users/{id}/` - 删除
- `GET /users/active/` - 自定义 action
- `POST /users/{id}/deactivate/` - 自定义 action

## 🔐 认证系统

FastAPI-CBV 提供了多种认证方式：

### Token 认证

```python
from fastapi_cbv import TokenAuthentication, APIView
from fastapi import HTTPException, status

class MyTokenAuth(TokenAuthentication):
    """自定义 Token 认证"""
    
    async def authenticate_credentials(self, token: str):
        # 验证 token 并返回用户
        user = await verify_token(token)  # 你的验证逻辑
        if not user:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid token"
            )
        return (user, token)

class ProtectedView(APIView):
    authentication_classes = [MyTokenAuth]
    
    async def get(self):
        user = self.request.state.user
        return {"message": f"Hello {user.username}!"}
```

### Bearer/JWT 认证

```python
from fastapi_cbv import BearerAuthentication
import jwt

class JWTAuthentication(BearerAuthentication):
    """JWT 认证"""
    
    async def authenticate_credentials(self, token: str):
        try:
            payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
            user = await User.get(id=payload["user_id"])
            return (user, token)
        except jwt.ExpiredSignatureError:
            raise HTTPException(401, "Token expired")
        except jwt.InvalidTokenError:
            raise HTTPException(401, "Invalid token")
```

### API Key 认证

```python
from fastapi_cbv import APIKeyAuthentication

class MyAPIKeyAuth(APIKeyAuthentication):
    """API Key 认证 (支持 Header 或 Query 参数)"""
    
    api_key_header = 'X-API-Key'      # Header: X-API-Key: your-key
    api_key_query_param = 'api_key'   # Query: ?api_key=your-key
    
    async def authenticate_credentials(self, api_key: str):
        # 验证 API Key
        if api_key == "valid-api-key":
            return ({"api_key": api_key}, api_key)
        raise HTTPException(401, "Invalid API Key")
```

### Basic 认证

```python
from fastapi_cbv import BasicAuthentication

class MyBasicAuth(BasicAuthentication):
    """HTTP Basic 认证"""
    
    async def authenticate_credentials(self, username: str, password: str, request):
        user = await User.get_or_none(username=username)
        if user and verify_password(password, user.password_hash):
            return (user, None)
        raise HTTPException(401, "Invalid credentials")
```

## 🛡️ 权限控制

### 内置权限类

```python
from fastapi_cbv import (
    AllowAny,              # 允许所有访问
    IsAuthenticated,       # 仅允许已认证用户
    IsAdminUser,           # 仅允许管理员
    IsAuthenticatedOrReadOnly,  # 已认证用户或只读
    IsOwnerOrReadOnly,     # 对象所有者或只读
)

class UserViewSet(ModelViewSet):
    permission_classes = [IsAuthenticated]
    
    def get_queryset(self):
        return User.all()

class AdminOnlyView(APIView):
    permission_classes = [IsAdminUser]
    
    async def get(self):
        return {"message": "Admin area"}
```

### 自定义权限类

```python
from fastapi_cbv import BasePermission
from fastapi import Request

class IsPremiumUser(BasePermission):
    """仅允许付费用户访问"""
    
    def has_permission(self, request: Request, view) -> bool:
        user = getattr(request.state, 'user', None)
        if user is None:
            return False
        return getattr(user, 'is_premium', False)
    
    def has_object_permission(self, request: Request, view, obj) -> bool:
        # 对象级别的权限检查
        return self.has_permission(request, view)

class PremiumContentView(APIView):
    permission_classes = [IsAuthenticated, IsPremiumUser]
    
    async def get(self):
        return {"content": "Premium content here"}
```

## ⚠️ 异常处理

### 内置异常类

```python
from fastapi_cbv import (
    # 4xx 客户端错误
    ValidationError,       # 400 验证失败
    ParseError,            # 400 解析错误
    AuthenticationFailed,  # 401 认证失败
    NotAuthenticated,      # 401 未认证
    PermissionDenied,      # 403 权限不足
    NotFound,              # 404 未找到
    MethodNotAllowed,      # 405 方法不允许
    Conflict,              # 409 冲突
    Throttled,             # 429 请求过多
    
    # 5xx 服务器错误
    ServerError,           # 500 服务器错误
    ServiceUnavailable,    # 503 服务不可用
)

class UserDetailView(RetrieveUpdateDestroyAPIView):
    async def get_object(self):
        user = await User.get_or_none(id=self.kwargs.get('id'))
        if not user:
            raise NotFound("User not found")
        return user
```

### 设置全局异常处理

```python
from fastapi import FastAPI
from fastapi_cbv import setup_exception_handlers, ExceptionHandlerMiddleware

app = FastAPI()

# 方式 1: 使用便捷函数注册所有异常处理器
setup_exception_handlers(app)

# 方式 2: 使用中间件 (支持 debug 模式)
app.add_middleware(ExceptionHandlerMiddleware, debug=True)
```

异常响应格式：
```json
{
    "detail": "User not found",
    "code": "not_found"
}
```

## 📊 分页

```python
from fastapi_cbv import TortoisePagination, ModelViewSet

class UserViewSet(ModelViewSet):
    pagination_class = TortoisePagination
    # 默认每页 10 条，可通过 ?page_size=20 自定义
    
    def get_queryset(self):
        return User.all()
```

分页响应格式：
```json
{
    "count": 100,
    "next": "/api/users/?page=2",
    "previous": null,
    "results": [...]
}
```

查询参数：
- `?page=1` - 页码
- `?page_size=20` - 每页数量

## 🔍 过滤和搜索

```python
from fastapi_cbv import (
    TortoiseFilterBackend,
    TortoiseSearchBackend,
    TortoiseOrderingBackend,
    ModelViewSet
)

class PostViewSet(ModelViewSet):
    filter_backends = [
        TortoiseFilterBackend,    # 字段过滤
        TortoiseSearchBackend,    # 全文搜索
        TortoiseOrderingBackend   # 排序
    ]
    search_fields = ['title', 'content']
    ordering_fields = ['created_at', 'title', 'views']
    ordering = ['-created_at']  # 默认排序
    
    def get_queryset(self):
        return Post.all()
```

支持的查询参数：
- `?search=keyword` - 全文搜索（在 search_fields 中搜索）
- `?ordering=created_at` - 升序排序
- `?ordering=-created_at` - 降序排序
- `?title__icontains=hello` - 字段过滤（支持 Tortoise ORM 查询语法）
- `?author_id=1` - 精确匹配过滤

## 🔧 Mixin 组合

灵活组合 Mixin 创建自定义视图：

```python
from fastapi_cbv import (
    GenericAPIView,
    CreateModelMixin,
    ListModelMixin,
    RetrieveModelMixin,
    UpdateModelMixin,
    DestroyModelMixin
)

# 只读视图（只支持列表和详情）
class ReadOnlyView(ListModelMixin, RetrieveModelMixin, GenericAPIView):
    def get_queryset(self):
        return User.all()

# 只支持创建和删除
class CreateDeleteView(CreateModelMixin, DestroyModelMixin, GenericAPIView):
    def get_queryset(self):
        return User.all()

# 自定义逻辑
class CustomView(CreateModelMixin, ListModelMixin, GenericAPIView):
    async def get(self, **kwargs):
        # 添加自定义逻辑
        return await self.list(**kwargs)
    
    async def post(self, **kwargs):
        result = await self.create(**kwargs)
        # 创建后发送通知
        await send_notification(result)
        return result
```

## ⚙️ 默认配置

FastAPI-CBV 提供了开箱即用的默认配置：

```python
class UserDetailView(RetrieveUpdateDestroyAPIView):
    serializer_class = UserSerializer
    # 以下配置已经是默认值，无需重复定义：
    # lookup_field = "id"                    # 默认使用 id 字段
    # datetime_format = "%Y-%m-%d %H:%M:%S"  # 默认日期时间格式
    # date_format = "%Y-%m-%d"               # 默认日期格式
```

**自定义覆盖：**
```python
class CustomView(RetrieveUpdateDestroyAPIView):
    serializer_class = UserSerializer
    lookup_field = "username"  # 使用 username 代替 id
    datetime_format = "%d/%m/%Y %H:%M"  # 欧洲格式
```

## 📁 项目结构

```
fastapi-cbv/
├── fastapi_cbv/
│   ├── __init__.py           # 导出所有公共 API
│   ├── views/
│   │   ├── base.py           # APIView, GenericAPIView
│   │   ├── mixins.py         # CRUD Mixin 类
│   │   ├── generics.py       # 通用视图类
│   │   └── viewsets.py       # ViewSet 类
│   ├── decorators.py         # @cbv, @action 装饰器
│   ├── routers.py            # CBVRouter
│   ├── authentication.py     # 认证类
│   ├── permissions.py        # 权限类
│   ├── exceptions.py         # 异常类和处理器
│   └── tortoise_integration.py  # Tortoise ORM 集成
├── examples/                 # 示例代码
├── tests/                    # 测试用例
└── README.md
```

## 📋 与 Django REST Framework 对比

| Django REST Framework | FastAPI CBV | 说明 |
|----------------------|-------------|------|
| `APIView` | `APIView` | 基础类视图 |
| `GenericAPIView` | `GenericAPIView` | 通用视图基类 |
| `ListCreateAPIView` | `ListCreateAPIView` | 列表 + 创建 |
| `RetrieveUpdateDestroyAPIView` | `RetrieveUpdateDestroyAPIView` | 详情 + 更新 + 删除 |
| `ModelViewSet` | `ModelViewSet` | 完整 CRUD ViewSet |
| `ReadOnlyModelViewSet` | `ReadOnlyModelViewSet` | 只读 ViewSet |
| `@action` | `@action` | 自定义 action 装饰器 |
| `@api_view` | `@cbv` | 类视图装饰器 |
| `serializers.ModelSerializer` | `create_tortoise_serializer()` | 模型序列化器 |
| `IsAuthenticated` | `IsAuthenticated` | 权限类 |
| `BaseAuthentication` | `BaseAuthentication` | 认证基类 |
| `APIException` | `APIException` | 异常基类 |

## 📚 完整示例

查看 `examples/complete_example.py` 了解完整的使用示例，包括：
- 模型定义
- 自动序列化器生成
- 各种类型的视图
- 认证和权限配置
- 路由注册
- 异常处理
- 过滤和分页

运行示例：
```bash
cd examples
python complete_example.py
# 访问 http://localhost:8000/docs 查看 API 文档
```

## 🧪 测试

```bash
# 运行所有测试
pytest tests/ -v

# 运行并查看覆盖率
pytest tests/ --cov=fastapi_cbv --cov-report=html
```

## 🤝 贡献

欢迎贡献代码！请查看贡献指南了解详情。

## 📄 许可证

MIT License