Django REST Framework style class-based views for FastAPI with Tortoise ORM support

These details have not been verified by PyPI

Project links

Project description

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

Project details

These details have not been verified by PyPI

Project links

Release history Release notifications | RSS feed

This version

0.2.0

Nov 28, 2025

0.1.0

Oct 4, 2025

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distributions

No source distribution files available for this release.See tutorial on generating distribution archives.

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

The dropdown lists show the available interpreters, ABIs, and platforms. Enable javascript to be able to filter the list of wheel files.

fastapi_cbv-0.2.0-py3-none-any.whl (36.3 kB view details)

Uploaded Nov 28, 2025 Python 3

File details

Details for the file fastapi_cbv-0.2.0-py3-none-any.whl.

File metadata

Download URL: fastapi_cbv-0.2.0-py3-none-any.whl
Upload date: Nov 28, 2025
Size: 36.3 kB
Tags: Python 3
Uploaded using Trusted Publishing? No
Uploaded via: twine/6.2.0 CPython/3.13.5

File hashes

Hashes for fastapi_cbv-0.2.0-py3-none-any.whl
Algorithm	Hash digest
SHA256	`19c87adc5e2d7229cf460ad9523477bdc64559617a1f35b6561190fbe5586a6c`
MD5	`54eeeae8122f98a7e37584759b699b77`
BLAKE2b-256	`152d4a7cd952bb044920c3b59b40d936d2a078656c3836f5b91714f43d4b0748`

See more details on using hashes here.

fastapi-cbv 0.2.0

Navigation

Verified details

Maintainers

Unverified details

Project links

Meta

Classifiers

Project description

FastAPI CBV (Class-Based Views)

✨ 特性

📦 安装

🚀 快速开始

1. 基础 APIView

2. 模型 CRUD 操作

3. ViewSet 用法

🔐 认证系统

Token 认证

Bearer/JWT 认证

API Key 认证

Basic 认证

🛡️ 权限控制

内置权限类

自定义权限类

⚠️ 异常处理

内置异常类

设置全局异常处理

📊 分页

🔍 过滤和搜索

🔧 Mixin 组合

⚙️ 默认配置

📁 项目结构

📋 与 Django REST Framework 对比

📚 完整示例

🧪 测试

🤝 贡献

📄 许可证

Project details

Verified details

Maintainers

Unverified details

Project links

Meta

Classifiers

Release history Release notifications | RSS feed

Download files

Source Distributions

Built Distribution

File details

File metadata

File hashes