FastAPI CBV (Class-Based Views)

基于 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

📦 安装

pip install fastapi-cbv

依赖要求：

Python >= 3.8
FastAPI >= 0.68.0
Tortoise ORM >= 0.19.0 (可选，用于 ORM 集成)

🚀 快速开始

1. 基础 APIView

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 操作

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 用法

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 认证

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 认证

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 认证

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 认证

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")

🛡️ 权限控制

内置权限类

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"}

自定义权限类

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"}

⚠️ 异常处理

内置异常类

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

设置全局异常处理

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)

异常响应格式：

{
    "detail": "User not found",
    "code": "not_found"
}

📊 分页

from fastapi_cbv import TortoisePagination, ModelViewSet

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

分页响应格式：

{
    "count": 100,
    "next": "/api/users/?page=2",
    "previous": null,
    "results": [...]
}

查询参数：

?page=1 - 页码
?page_size=20 - 每页数量

🔍 过滤和搜索

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 创建自定义视图：

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 提供了开箱即用的默认配置：

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

自定义覆盖：

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 了解完整的使用示例，包括：

模型定义
自动序列化器生成
各种类型的视图
认证和权限配置
路由注册
异常处理
过滤和分页

运行示例：

cd examples
python complete_example.py
# 访问 http://localhost:8000/docs 查看 API 文档

🧪 测试

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

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

🤝 贡献

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

📄 许可证

MIT License

Name		Name	Last commit message	Last commit date
Latest commit History 6 Commits
.github/workflows		.github/workflows
.history		.history
examples		examples
fastapi_cbv		fastapi_cbv
scripts		scripts
tests		tests
.gitignore		.gitignore
CHANGELOG.md		CHANGELOG.md
DEFAULT_CONFIG_UPDATE.md		DEFAULT_CONFIG_UPDATE.md
HOW_TO_PUBLISH.md		HOW_TO_PUBLISH.md
LICENSE		LICENSE
Makefile		Makefile
PUBLISH_GUIDE.md		PUBLISH_GUIDE.md
PYPI_GUIDE.md		PYPI_GUIDE.md
QUICK_REFERENCE.md		QUICK_REFERENCE.md
README.md		README.md
RELEASE_READY.md		RELEASE_READY.md
publish_to_pypi.py		publish_to_pypi.py
pyproject.toml		pyproject.toml
release.sh		release.sh
test.db-shm		test.db-shm
test.db-wal		test.db-wal
uv.lock		uv.lock

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Repository files navigation

FastAPI CBV (Class-Based Views)

✨ 特性

📦 安装

🚀 快速开始

1. 基础 APIView

2. 模型 CRUD 操作

3. ViewSet 用法

🔐 认证系统

Token 认证

Bearer/JWT 认证

API Key 认证

Basic 认证

🛡️ 权限控制

内置权限类

自定义权限类

⚠️ 异常处理

内置异常类

设置全局异常处理

📊 分页

🔍 过滤和搜索

🔧 Mixin 组合

⚙️ 默认配置

📁 项目结构

📋 与 Django REST Framework 对比

📚 完整示例

🧪 测试

🤝 贡献

📄 许可证

About

Uh oh!

Releases

Packages

Languages

License

miaokela/fastapi-cbv

Folders and files

Latest commit

History

Repository files navigation

FastAPI CBV (Class-Based Views)

✨ 特性

📦 安装

🚀 快速开始

1. 基础 APIView

2. 模型 CRUD 操作

3. ViewSet 用法

🔐 认证系统

Token 认证

Bearer/JWT 认证

API Key 认证

Basic 认证

🛡️ 权限控制

内置权限类

自定义权限类

⚠️ 异常处理

内置异常类

设置全局异常处理

📊 分页

🔍 过滤和搜索

🔧 Mixin 组合

⚙️ 默认配置

📁 项目结构

📋 与 Django REST Framework 对比

📚 完整示例

🧪 测试

🤝 贡献

📄 许可证

About

Resources

License

Uh oh!

Stars

Watchers

Forks

Releases

Packages 0

Languages

Packages