from ninja import NinjaAPI
api = NinjaAPI()
@api.get('/hello')
def hello(request):
return {'hello': 'world'}
在url.py配置url路由
from django.contrib import admin
from django.urls import path
from .api import api
urlpatterns = [
path('admin/', admin.site.urls),
path('api/', api.urls),
]
@api.get("/weapons")
def list_weapons(request, limit: int = 10, offset: int = 0):
return weapons[offset:offset+limit]
位置参数和可选参数综合使用
@api.get("/weapons/search")
def search_weapons(request, keyword: str, offset: int = 0):
results = [w for w in weapons if keyword in w.lower()]
return results[offset: offset+10]
使用Schema定义
import datetime
from typing import List
from pydantic import Field
from ninja import Query, Schema
class Filters(Schema):
limit: int = 100
offset: int = None
query: str = None
category__in: List[str] = Field(None, alias="categories")
@api.get("/filter")
def events(request, filters: Filters = Query(...)):
return {"filters": filters.dict()}
注：Query()函数用来标记filters参数是查询参数。
请求体
使用Schema
from ninja import Schema
class Item(Schema):
name: str
description: str = None
price: float
quantity: int
@api.post("/items")
def create(request, item: Item):
return item
from ninja import Form
@api.post("/login")
def login(request, username: str = Form(...), password: str = Form(...)):
return {"username": username, "password": password}
使用Schema
from ninja import Schema, Form
class Item(Schema):
name: str
description: str = None
price: float
quantity: int
@api.post("/item")
def create(request, item: Item = Form(...)):
return item
路径参数，请求参数和表单
from ninja import Schema, Form
class Item(Schema):
name: str
description: str = None
price: float
quantity: int
@api.post("/items/{item_id}")
def update(request, item_id: int, q: str, item: Item=Form(...)):
return {"item_id": item_id, "item": item, "q": q}
将空表单值设置为默认值
from ninja import Schema, Form
from pydantic.fields import ModelField
from typing import Generic, TypeVar
PydanticField = TypeVar("PydanticField")
class EmptyStrToDefault(Generic[PydanticField]):
@classmethod
def __get_validators__(cls):
yield cls.validate
@classmethod
def validate(cls, value: PydanticField, field: ModelField) -> PydanticField:
if value == "":
return field.default
return value
class Item(Schema):
name: str
description: str = None
price: EmptyStrToDefault[float] = 0.0
quantity: EmptyStrToDefault[int] = 0
in_stock: EmptyStrToDefault[bool] = True
@api.post("/item-blank-default")
def update(request, item: Item=Form(...)):
return item.dict()
文件上传
单个文件上传
from ninja import File
from ninja.files import UploadedFile
@api.post("/upload")
def upload(request, file: UploadedFile = File(...)):
data = file.read()
return {"name": file.name, "len": len(data)}
多个文件上传
from typing import List
from ninja import File
from ninja.files import UploadedFile
@api.post("/upload-many")
def upload_many(request, files: List[UploadedFile] = File(...)):
return [f.name for f in files]
from ninja import Schema
class UserIn(Schema):
username: str
password: str
class UserOut(Schema):
id: int
username: str
@api.post("/users/", response=UserOut)
def create_user(request, data: UserIn):
user = User(username=data.username)
user.set_password(data.pasword)
user.save()
return user
注：响应体Schema会限制返回数据，仅仅返回定义在Schema中的数据。
返回体为嵌套对象
# model.py
from django.db import models
class Task(models.Model):
title = models.CharField(max_length=200)
is_completed = models.BooleanFied(default=False)
owner = models.ForeignKey("auth.User", null=True, blank=True)
# api.py
from typing import List
from ninja import Schema
class UserSchema(Schema):
id: int
first_name: str
last_name: str
class TaskSchema(Schema):
id: int
title: str
is_completed: bool
owner: UserSchema = None # None -> to mark it as optional
@api.get("/tasks", response=List[TaskSchema])
def tasks(request):
queryset = Task.objects.all()
return list(queryset) # or return queryset
返回文件对象或图片
文件或图片的schema均返回地址。
# model.py
class Picture(models.Model):
title = models.CharField(max_length=100)
image = models.ImageField(upload_to="images")
# api.py
class PictureSchema(Schema):
title: str
image: str
返回状态码和数据
from ninja import Schema
class Token(Schema):
token: str
expires: date
class Message(Schema):
message: str
@api.post("/login", response={200: Token, 401:Message, 402:Message})
def login(request, payload: Auth):
if auth_not_valid:
return 401, {"message": "Unauthorized"}
if negative_balance:
return 402, {"message": "xxxx"}
return 200, {"token": xx, ....}
当返回状态码和数据一致时，可使用4xx
from ninja.response import codes_1xx
from ninja.response import codes_2xx
from ninja.response import codes_3xx
from ninja.response import codes_4xx
from ninja.response import codes_5xx
@api.post('/login', response={200: Token, codes_4xx: Message})
def login(request, payload: Auth):
if auth_not_valid:
return 401, {'message': 'Unauthorized'}
if negative_balance:
return 402, {'message': 'Insufficient balance amount. Please proceed to a payment page.'}
return 200, {'token': xxx, ...}
自我套用
class Organization(Schema):
title: str
part_of: 'Organization' = None
Organization.update_forward_refs() # this is important
Model的Schema
选择包含字段
from django.contrib.auth.models import User
from ninjia import ModelSchema
class UserSchema(ModelSchema):
class Config:
model = User
model_fields = ["id", "username", "first_name", "last_name"]
选择不包含字段
from django.contrib.auth.models import User
from ninja import ModelSchema
class UserSchema(ModelSchema):
class Config:
model = User
model_exclude = ["password", "last_login", "user_permissions"]
覆盖字段(修改某些字段注释或者添加新字段)
class GroupSchema(ModelSchema):
class Config:
model = Group
model_fields = ["id", "name"]
class UserSchema(ModelSchema):
groups: List[GroupSchema] = []
class Config:
model = User
model_fields = ["id", "username", "first_name", "last_name"]
其他,使用create_schema
语法
def create_schema(
model, # django model
name = "", # name for the genarated class ,if empty model name is used
depth = 0, # if>0 schema will be also created for nested ForeignKeys and Many2Many (with the provided depth of lookup)
fields: list[str] = None,
exclude: list[str] = None,
custom_fields: list[typle[str, Any, Any]] = None # if passed - this will override default field types (or add new fields)
)
from ninja import NinjaAPI
from ninja.security import django_auth
api = NinjiaAPI(csrf=True)
@api.get("/pets", auth=django_auth)
def pets(request):
return "Authenticated user {}".format(request.auth)
from ninja import NinjaAPI, Form
from ninja.security import HttpBearer
class GlobalAuth(HttpBearer):
def authenticate(self, request, token):
if token == "supersecret":
return token
api = NinjaAPI(auth=GlobalAuth())
在全局认证时，如果某些函数不需要全局认证，则将该路由路径中的auth设置为None。
from ninja import NinjaAPI, Form
from ninja.security import HttpBearer
class GlobalAuth(HttpBearer):
def authenticate(self ,request, token):
if token == "supersecret":
return token
api = NinjaAPI(auth=GlobalAuth())
@api.post("/token", auth=None) # overriding global auth
def get_token(request, username: str = Form(...), password: str = Form(...)):
if username == "admin" and password == "password":
return {"token": "supersecret"}
方法三、路由器认证
api.add_router("/events", events_router, auth=BasicAuth())
# or we can write as this
# router = Router(auth=BasicAuth())
from ninja import NinjaAPI
class MySuperApi(NinjaAPI):
def get_openapi_operation_id(self, operation):
# here you can access operation ( .path , .view_func, etc)
return ...
api = MySuperApi()
@api.get()
...
@api.post("/tasks", url_name="tasks")
def xxx_operation(request):
pass
# then you can get the url with
reverse("api-1.0.0: tasks")
版本控制
使用Django Ninja，可以轻松地从单个 Django 项目运行多个 API 版本。
不同的API版本号
# api_v1.py
from ninja import NinjaAPI
api = NinjaAPI(version="1.0.0")
@api.get("/hello")
def hello(request):
return {"message": "Hello form v1"}
# api_v2.py
from ninja import NinjaAPI
api = NinjaAPI(version="2.0.0")
@api.get("/hello")
def hello(reqeust):
return {"message": "Hello from v2"}
# urls.py
from api_v1 import api as api_v1
from api_v2 import api as api_v2
urlpatterns = [
...
path("api/v1/", api_v1.urls),
path("api/v2/", api_v2.urls),
]
在大多数情况下，REST API 的默认内容类型是 JSON，但如果您需要使用其他内容类型（如 YAML、XML、CSV）或使用更快的 JSON 解析器，Django Ninja提供了parser配置。
YAML解析器
import yaml
from typing import List
from ninja import NinjaAPI, Schema
from ninja.parser import Parser
class MyYamlParser(Parser):
def parse_body(self, request):
return yaml.safe_load(request.body)
api = NinjaAPI(parser=MyYamlParser())
class Payload(Schema):
ints: List[int]
string: str
f: float
@api.post("/yaml")
def operation(request, payload: Payload):
return payload.dict()
ORJSON 解析器
import orjson
from ninja import NinjaAPI
from ninja.parser import Parser
class ORJSONParser(Parser):
def parse_body(self, request):
return orjson.loads(request.body)
api = NinjaAPI(parser=ORJSONParser())
响应渲染器
REST API 最常见的响应类型通常是JSON。Django Ninja还支持自定义渲染器，这可以让您灵活地设计自己的媒体类型
创建渲染器
from ninja import NinjaAPI
from ninja.renderers import BaseRender
class MyRenderer(BaseRender):
media_type = "text/plain"
def render(self, request, data, *, response_status):
return ... # your serialization here
api = NinjaAPI(renderer=MyRenderer())
from ninja import NinjaAPI
api = NinjaAPI(csrf=True)
将API与基于cookie的身份验证一起使用是不安全！
# 如果这样做，会报错。
from ninja import NinjaAPI
from ninja.security import django_auth
api = NinjaAPI(auth=django_auth)
# 必须启用csrf检查，基于cokkie的身份验证才安全。
from ninja import NinjaAPI
from ninja.security import django_auth
api = NinjaAPI(auth=django_auth, csrf=True)