ninja 轻量级接口开发框架
Django Ninja是一个为Django设计的轻量级现代REST API框架,作者是乌克兰程序员Vitaliy Kucheryaviy。Django Ninja的灵感来源于FastAPI,Django Ninja框架集成了Pydantic,能够优雅的实现数据验证、序列化和自动生成Swagger文档,与学习曲线陡峭的“重量级”框架Django DRF相比,Django Ninja具有轻量级、开发体验友好、心智负担极低的特点。如果你喜欢简洁的FastAPI而非复杂的Django DRF但又不愿抛弃Django ORM,那Django Ninja绝对是你的不二之选。
官方网站:https://django-ninja.dev/
Github:https://github.com/vitalik/django-ninja
安装Django Ninja
在Django工程中,执行以下命令安装Django Ninja。
pip install django-ninja
框架安装完成后,在Django配置文件的INSTALLED_APPS中添加相关配置。
settings.py
INSTALLED_APPS = [
# ... other apps
'ninja',
]
注意实际上Django Ninja的核心功能并不需要在INSTALLED_APPS中添加ninja配置,这里添加配置的唯一作用是能让Swagger UI(或Redoc)的静态资源文件从本地staticfiles加载而非从网络CDN加载(默认CDN可能为cdn.jsdelivr.net),如果你的开发环境能正常访问国际网络那么其实是不需要配置它的。
Django Ninja简单使用
我们这里创建一个最简单的Django工程,目录结构如下,其中api.py是我们新添加的Python源码文件,其中包含API接口的实现代码。
blog
|_ blog # Django APP
|_ api.py # Django Ninja的API接口实现
|_ wsgi.py
|_ urls.py
|_ settings.py # 工程配置文件
|_ manage.py
api.py
from ninja import NinjaAPI
api = NinjaAPI()
@api.get('/hello')
def hello(request):
return {'code': 0, 'message': 'Hello World'}
代码非常简单,我们定义了一个叫/hello的GET接口,它返回一个dict类型的数据,Django Ninja会将其序列化为JSON返回给客户端。api是NinjaAPI的实例,NinjaAPI是Django Ninja的核心类,在这里它的作用是API路由管理器,我们可以基于它使用@api.get()、@api.post()注册API端点,以及生成Django所需的URL配置。
编写好API接口函数实现后,我们还需要将其注册到Django的路由系统中,这需要用到api.urls属性,我们这里将其挂载到/api路径下。
urls.py
from django.contrib import admin
from django.urls import path
from blog.api import api
urlpatterns = [
path('admin/', admin.site.urls),
path('api/', api.urls),
]
此时启动工程,我们可以使用浏览器访问http://localhost:8000/api/hello,查看响应结果。
Pydantic模型简介
Django Ninja使用Pydantic定义数据模型,数据的字段验证、序列化等都依赖Pydantic模型定义。我们这里继续之前的例子,添加一个schemas.py用于定义Pydantic数据模型。
blog
|_ blog # Django APP
|_ api.py # Django Ninja的API接口实现
|_ schemas.py # 数据模型定义
|_ wsgi.py
|_ urls.py
|_ settings.py # 工程配置文件
|_ manage.py
Pydantic数据模型需要继承Django Ninja的Schema类,这里我们定义一个用户模型,包含ID、名字、邮箱等字段,其中is_active有默认值True,而birthday是一个可为None的日期类型。
schemas.py
from ninja import Schema
from datetime import date
class UserSchema(Schema):
id: int
username: str
email: str
is_active: bool = True
birthday: date | None = None
接口实现中,我们编写一个POST接口,其中使用请求体接收我们的UserSchema数据模型并将受到的信息打印到控制台上。
api.py
from ninja import NinjaAPI
from blog.schemas import UserSchema
api = NinjaAPI()
@api.post('/add-user')
def add_user(request, data: UserSchema):
print(data.id)
print(data.username)
print(data.email)
print(data.is_active)
print(data.birthday)
return {'code': 0, 'message': 'success'}
启动工程后,我们可以使用Postman等调试工具访问POST http://localhost:8000/api/add-user,如果我们的请求参数中缺少Pydantic数据模型定义的必填项,会发现服务端自动返回422 Unprocessable Entity,并给出了错误提示。有关如何定制错误处理逻辑,将在后文介绍,如果你对Pydantic并不熟悉,可以参考Python/第三方库/pydantic-数据模型验证库章节。
处理HTTP请求
处理HTTP方法
Django Ninja中针对HTTP的GET、POST、PUT、DELETE、PATCH都定义了对应的装饰器,我们需要使用这些装饰器修饰API接口函数。
@api.get("/items")
def my_view(request):
pass
@api.post("/items")
def my_view(request):
pass
@api.put("/items/{id}")
def my_view(request):
pass
@api.delete("/items/{id}")
def my_view(request):
pass
读取URL参数
Django Ninja中,user_id: int = Query(...)表达声明了一个user_id参数,它是一个数字类型的URL参数。Query是一个辅助的参数标记类,虽然使用的是Python的默认值语法,但在这里它的语义实际是一个特殊的标注工具,Query也是基于Pydantic实现的,用于显式声明函数参数来自查询参数。
from ninja import NinjaAPI, Query
api = NinjaAPI()
@api.get('/demo')
def demo(request, user_id: int = Query(...)):
print(user_id)
return {'code': 0, 'message': 'success'}
注:Query中的...是Python的内置Ellipsis对象,在这里它表达该参数是必填项,如果我们传的参数类型错误或缺少参数,框架会返回HTTP 422 Unprocessable Entity错误,并提示参数缺失。
Query还支持许多额外的元数据配置,常用的包括:
default:默认值,如果是...表示必填参数,如果是None或其它值,则代表可选参数alias:参数的别名title:参数标题,会显示在Swagger UI中,指定该参数主要用于提升Swagger UI接口文档可读性description:参数描述,作用同titlegt/ge/lt/le:用于数值类型的参数约束min_length/max_length:用于字符串类型的字符串长度约束regex:用于字符串类型的正则表达式约束
实际上,如果没有额外的配置项,Django Ninja支持更简化的写法,在API接口函数中定义的基本类型参数(int、str、float、bool等)默认就会从URL参数中解析。
from ninja import NinjaAPI
api = NinjaAPI()
@api.get('/demo')
def demo(request, user_id: int):
print(user_id)
return {'code': 0, 'message': 'success'}
如果我们想将user_id设置为一个可选参数,可以为其指定一个默认值。
from ninja import NinjaAPI
api = NinjaAPI()
@api.get('/demo')
def demo(request, user_id: int | None = None):
print(user_id)
return {'code': 0, 'message': 'success'}
读取路径参数
下面代码展示了如何处理路径参数,我们需要在路由字符串中设置一个占位符,并在函数参数中添加对应的参数名并标注类型。
from ninja import NinjaAPI
api = NinjaAPI()
@api.get('/demo/{user_id}')
def demo(request, user_id: int):
print(user_id)
return {'code': 0, 'message': 'success'}
路径参数默认都是必填的。
读取请求头
Django中读取请求头非常简单,我们直接使用视图函数上的request对象的request.headers属性即可实现。
from ninja import NinjaAPI
api = NinjaAPI()
@api.get('/demo')
def demo(request):
ua = request.headers.get('User-Agent')
print(ua)
return {'code': 0, 'message': 'success'}
但如果我们需要请求头在Swagger UI中显示并自动验证、解析,或是添加一些额外的元数据配置,此时可以使用Django Ninja提供的Header参数标记类。
from ninja import NinjaAPI, Header
api = NinjaAPI()
@api.get('/demo')
def demo(request, x_api_key: str = Header(...)):
print(x_api_key)
return {'code': 0, 'message': 'success'}
读取请求体
读取请求体前面我们已经演示过了,请求体的参数结构需要声明为继承Django Ninja的Schema的派生类,Django Ninja会自动解析JSON数据。
from ninja import NinjaAPI, Schema
api = NinjaAPI()
class ItemSchema(Schema):
name: str
price: float
@api.post('/demo')
def demo(request, item: ItemSchema):
print(item.name)
print(item.price)
return {'code': 0, 'message': 'success'}
对于请求体参数的校验,我们可以使用Field标记类。
class ItemSchema(Schema):
name: str = Field(..., min_length=3)
price: float = Field(..., gt=0)
当参数校验失败时,服务端也会返回422 Unprocessable Entity。
文件上传
对于文件上传,我们可以在API接口函数中使用UploadedFile类型的参数,同时使用Django Ninja提供的File标记类。下面代码中我们实现了一个最简单的文件上传,并打印文件的bytes类型二进制数据。
from ninja import NinjaAPI, UploadedFile, File
api = NinjaAPI()
@api.post('/demo')
def demo(request, file: UploadedFile = File(...)):
data = file.read()
print(data)
return {'code': 0, 'message': 'success'}
返回HTTP响应
返回JSON
Django Ninja中,返回字典、列表或使用Pydantic模型创建的数据实例都会被自动序列化为JSON输出。
from ninja import NinjaAPI, Schema
api = NinjaAPI()
class ItemSchema(Schema):
name: str
price: float
@api.get('/demo')
def demo(request):
item = ItemSchema(name='袜子', price=10.0)
return {'code': 0, 'message': 'success', 'data': item}
返回HTTP状态码
Django Ninja支持API接口函数返回元组,其中第一个元素是HTTP状态码,第二个元素才是真正的返回数据,不过这种写法还需要在操作装饰器(例如@api.get())上指定response参数,其中定义了可能返回的响应HTTP状态码和返回类型,Django Ninja会根据状态码选择对应的类型进行序列化和验证。
from ninja import NinjaAPI, Schema
api = NinjaAPI()
class ItemSchema(Schema):
name: str
price: float
@api.get('/demo', response={201: dict})
def demo(request):
item = ItemSchema(name='袜子', price=10.0)
return 201, {'code': 0, 'message': 'success', 'data': item}
返回数据模型
Django Ninja中返回从ORM中查询的结果时,最基础的方式是将结果转为dict或list,然后手动填充Pydantic模型实例,不过如果我们使用了操作装饰器的response参数,这一过程可以自动完成。Django Ninja提供了ModelSchema类,它可以直接基于已有的Django数据模型自动生成Pydantic模型,下面代码演示了相关用法。
from ninja import NinjaAPI, ModelSchema
from blog.models import Student
api = NinjaAPI()
class StudentSchema(ModelSchema):
class Meta:
model = Student
fields = '__all__'
@api.get('/demo', response=StudentSchema)
def demo(request):
student = Student.objects.filter(pk=1).first()
return student
注:上面代码其实有个小瑕疵,如果数据库中没有id为1的数据记录,但Django Ninja仍被指定为响应类型为StudentSchema,此时查询结果student实际是None,服务端就会报错,返回HTTP 500状态。如果我们的返回值可能为None,可以声明返回类型为StudentSchema | None。
Pydantic模型支持嵌套声明,下面例子中ApiResultSchema是一个“泛型”的通用响应体,其中的data字段才是具体的返回数据,我们将其声明为泛型类型T,同时在API接口函数的response配置中声明返回的Schema为ApiResultSchema[list[StudentSchema]]。
from typing import TypeVar, Generic
from ninja import NinjaAPI, Schema, ModelSchema
from blog.models import Student
api = NinjaAPI()
T = TypeVar('T')
class ApiResultSchema(Schema, Generic[T]):
code: int
message: str
data: T
class StudentSchema(ModelSchema):
class Meta:
model = Student
fields = '__all__'
@api.get('/demo', response=ApiResultSchema[list[StudentSchema]])
def demo(request):
students = Student.objects.all()
return ApiResultSchema(code=0, message='success', data=students)
注:ApiResultSchema[list[StudentSchema]]实际上是一个Pydantic“类型”类,ApiResultSchema[]中的方括号是Python中的泛型语法,方括号内的list[StudentSchema]是实际类型。
子路由划分
实际开发中,我们通常需要对路由系统按照功能模块进行合理的拆分,避免所有路由配置挂载到一个巨大的难以维护的文件中。我们这里准备下面的包目录结构,最外层的api.py是该模块的主入口,但功能被实际划分为了users和posts两个包。
|_ api.py
|_ users
|_ api.py
|_ posts
|_ api.py
Django Ninja中,Router类用于将路由模块化。Router实例支持的操作与NinjaAPI实例类似,都可以声明路由装饰器以及指定配置参数,但Router实例还可以挂载到另一个Router实例或NinjaAPI实例,形成子路由划分。
api.py
from ninja import NinjaAPI
from blog.users.api import router as users_router
from blog.posts.api import router as posts_router
api = NinjaAPI()
api.add_router('/users', users_router)
api.add_router('/posts', posts_router)
users/api.py
from ninja import Router
router = Router()
@router.get("/demo")
def demo(request):
return {'code': 0, 'message': 'success', 'data': 'users'}
posts/api.py
from ninja import Router
router = Router()
@router.get("/demo")
def demo(request):
return {'code': 0, 'message': 'success', 'data': 'posts'}
认证鉴权
APIKey认证
Django Ninja提供了一种内置的简易APIKey认证机制,它可以通过校验请求头或请求参数中的一个指定值来判断是否通过认证。下面例子中,我们全局设置了当前这组路由被APIKey认证机制保护,访问接口时如果没有正确传递X-API-Key请求头就会抛出HTTP 401错误。
from ninja import NinjaAPI
from ninja.errors import HttpError
from ninja.security import APIKeyHeader
class ApiKey(APIKeyHeader):
param_name = "X-API-Key"
def authenticate(self, request, key):
if key is None:
raise HttpError(401, 'Authentication required')
if key == 'abc123':
return {'key': key}
api = NinjaAPI(auth=ApiKey())
@api.get('/demo')
def demo(request):
return {'code': 0, 'message': 'success'}
我们的ApiKey类继承APIKeyHeader表达认证APIKey使用请求头方式传递,Django Ninja还支持APIKeyQuery使用请求参数传递APIKey。此外,前面代码我们是全局设置的APIKey认证校验,我们也可以针对单个API接口函数设置。
@api.get('/demo', auth=ApiKey())
def demo(request):
return {'code': 0, 'message': 'success'}
这里的auth参数也支持传递数组,组合多种认证方式。
集成Django Auth
Django Ninja也可以与Django内置的强大Django Auth模块集成使用,直接复用基于Session的一套登录认证和RBAC机制。下面例子我们的API接口函数被指定为必须登录(可以借助Django Admin界面或调用Django Auth的登陆函数)才能访问,否则会返回HTTP 401错误。
from ninja import NinjaAPI
from ninja.security import django_auth
api = NinjaAPI()
@api.get('/demo', auth=django_auth)
def demo(request):
return {'code': 0, 'message': 'success'}
除了django_auth,Django Ninja还提供了方便的django_auth_superuser,表示必须登录超级管理员账号才能访问。
from ninja.security import django_auth_superuser
登录后,对于权限的判断就和Django Auth一致了,我们可以访问request.user判断用户的认证和权限信息。下面代码用户除了必须登录,还必须拥有相关的权限(或是超级管理员)。
from ninja import NinjaAPI
from ninja.security import django_auth
api = NinjaAPI()
@api.get('/demo', auth=django_auth)
def demo(request):
if not request.user.has_perm('app.some_permission'):
return {'code': 403, 'message': 'Permission denied'}
return {'code': 0, 'message': 'success'}
全局异常处理
Django Ninja默认实现了一些异常处理逻辑,但有时默认行为不是我们想要的,Django Ninja也完全支持我们自定义框架的异常处理行为。
以请求参数校验异常为例,这种情况下框架会抛出ninja.errors.ValidationError,我们可以针对该类型统一设置所有的异常处理行为。下面代码中,我们使用@api.exception_handler(ValidationError)声明了该类型异常的处理函数,注意异常处理函数与普通的API接口函数不同,它的返回值需要是Django Ninja的响应对象或Django的HttpResponse或JsonResponse。
from ninja import NinjaAPI, Schema
from ninja.errors import ValidationError
api = NinjaAPI()
@api.exception_handler(ValidationError)
def custom_validation_error_handler(request, exc: ValidationError):
return api.create_response(request, {'code': 400, 'message': 'Validation error'}, status=200)
class ItemSchema(Schema):
name: str
price: float
@api.post('/demo')
def demo(request, item: ItemSchema):
print(item.name)
print(item.price)
return {'code': 0, 'message': 'success'}
此外,一般实际开发中,我们还会创建一个“兜底”的@api.exception_handler(Exception),如果特定异常类型有对应类型的异常处理函数会优先触发,否则触发兜底异常处理函数。
限流
Django Ninja自带了简单易用的限流功能,它基于Django自带的缓存框架实现。限流配置通常涉及AnonRateThrottle和AuthRateThrottle两个类,分别配置对匿名用户和登录用户的限流行为,下面是一个例子,代码中我们配置了匿名用户和登录用户调用接口都被限制为了最大每秒1次。
from ninja import NinjaAPI
from ninja.throttling import AnonRateThrottle, AuthRateThrottle
api = NinjaAPI()
@api.get('/demo', throttle=[AnonRateThrottle('1/s'), AuthRateThrottle('1/s')])
def demo(request):
return {'code': 0, 'message': 'success'}
限流声明的格式为requests/time-unit,例如10/s、20/m、100/5m等,时间单位支持s(秒)、m(分钟)、h(小时)、d(天),如果省略单位则默认为秒。
Swagger UI文档
Django Ninja默认会基于API接口函数定义生成Swagger UI文档,无需做任何特殊配置,我们直接使用浏览器访问/docs即可看到Swagger UI(如果NinjaAPI实例挂载在/api/下则是/api/docs)。尽管如此,实际开发中还是建议手动设置Swagger UI的开启和关闭,因为我们通常不希望Swagger UI在生产环境出现,下面代码例子展示了如何实现这一逻辑。
from ninja import NinjaAPI, Swagger
from blog import settings
docs = Swagger() if settings.DEBUG else None
api = NinjaAPI(docs=docs)