简介

Django Ninja 是一个 Web 框架，用于使用 Django 和 Python 3.6+ 类型提示构建 API。

主要特征：

简单： 设计易于使用且直观。
快速执行： 由于Pydantic和异步支持，性能非常高。
快速编码： 类型提示和自动文档让您只关注业务逻辑。
基于标准： 基于API的开放标准：OpenAPI（以前称为Swagger）和JSON Schema。
Django 友好：（显然）与 Django 核心和 ORM 具有良好的集成。
交互式 API 文档：（由OpenAPI / Swagger UI或Redoc提供）

启动项目

安装

1	pip install django-ninja

创建项目

1	django-admin startproject apidemo

基础配置

在urls.py

from django.contrib import admin
from django.urls import path
from ninja import NinjaAPI

api = NinjaAPI()


@api.get("/add")
def add(request, a: int, b: int):
    return {"result": a + b}


urlpatterns = [
    path("admin/", admin.site.urls),
    path("api/", api.urls),
]

启动项目

1	python manage.py runserver

打开交互式文档

浏览器打开：http://127.0.0.1:8000/api/docs
将看到自动的交互式 API 文档（由OpenAPI / Swagger UI或Redoc提供）：
docs接口图

请求数据

路径参数

路径参数 item_id 的值将作为参数 item_id 传递给您的函数。

# 不指定参数类型，默认为字符串
@api.get("/items/{item_id}")
def read_item(request, item_id):
    return {"item_id": item_id}

# 指定参数类型
@api.get("/items/{item_id}")
# 指定参数类型为int
def read_item(request, item_id: int):
    return {"item_id": item_id}

item_id会根据在接口输入的值返回：

1
2
3

{
  "item_id": "1"
}

也可进行路径参数转换：

# 路径参数指定int后前后不能有空格
@api.get("/items/{int:item_id}")
def read_item(request, item_id):
    return {"item_id": item_id}

查询参数

分页查询

weapons = ["Ninjato", "Shuriken", "Katana", "Kama", "Kunai", "Naginata", "Yari"]


@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, q: str, offset: int = 0):
    results = [w for w in weapons if q in w.lower()]
    print(q, results)
    return results[offset: offset + 10]

使用Schema

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 Router
from ninja import Schema


api = Router()

class Item(Schema):
    name: str
    description: str = None
    price: float
    quantity: int


@api.post("/items")
def create(request, item: Item):
    return item

=None作为默认值，说明这个字段可不传值为空

请求体+路径参数


@api.put("/items/{item_id}")
def update(request, item_id: int, item: Item):
    return {"item_id": item_id, "item": item.dict()}

请求体+路径参数+查询参数


@api.post("/items/{item_id}")
def update(request, item_id: int, item: Item, q: str):
    return {"item_id": item_id, "item": item.dict(), "q": q}

三者同时包含解释：

如果该参数也在路径中声明，它将被用作路径参数
如果参数是单数类型（如 int、float、str、bool 等），它将被解释为查询参数
如果参数声明为 Schema（或 Pydantic BaseModel）类型，它将被解释为请求体。

Form表单

Form表单数据作为参数

1
2
3

@api.post("/login")
def login(request, username: str = Form(...), password: str = Form(...)):
    return {'username': username, 'password': '*****'}

使用Schema

class Item(Schema):
    name: str
    description: str = None
    price: float
    quantity: int


@api.post("/items")
def create(request, item: Item = Form(...)):
    return item

表单+路径参数+查询参数

1
2
3

@api.post("/items/{item_id}")
def update(request, item_id: int, q: str, item: Item = Form(...)):
    return {"item_id": item_id, "item": item.dict(), "q": q}

可选表单空值设置

可选的表单字段通常以空值发送。该值被解释为空字符串，因此可能无法验证 int 或 bool 等字段。


from ninja import Form, Schema
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("/items-blank-default")
def update(request, item: Item=Form(...)):
    return item.dict()

文件上传

单文件上传

from ninja import NinjaAPI, 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 NinjaAPI, 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 NinjaAPI, Schema, UploadedFile, Form, File
from datetime import date

api = NinjaAPI()


class UserDetails(Schema):
    first_name: str
    last_name: str
    birthdate: date


@api.post('/user')
def create_user(request, details: UserDetails = Form(...), file: UploadedFile = File(...)):
    return [details.dict(), file.name]

UploadedFile是Django的UploadFile的一个别名，它拥有访问上传文件的所有方法和属性

read()
- 从文件中读取整个上传的文件。文件太大会报错。
multiple_chunks(chunk_size=None)
- 如果上传的文件足够大，需要分块读取，返回True。默认情况下是大于2.5M的文件。
chunks(chunk_size=None)
- 一个生成器，返回文件的块。
name
- 上传文件的文件名
size
- 上传文件的大小，以字节为单位
content_type
- 与文件一起上传的内容类型头
etc.

请求解析

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

如果像这样发送 YAML 作为请求正文:

ints:
 - 0
 - 1
string: hello
f: 3.14

它将被正确解析，并且有如下 JSON 输出:

{
  "ints": [
    0,
    1
  ],
  "string": "hello",
  "f": 3.14
}

ORJSON 解析器

orjson 是一个快速、准确的 Python JSON 库。它被评为最快的 JSON Python 库，并且比标准 json 库或其他第三方库更准确。

1 2	# 安装库 pip install orjson

import orjson
from ninja import NinjaAPI, Schema
from ninja.parser import Parser


class ORJSONParser(Parser):
    def parse_body(self, request):
        return orjson.loads(request.body)


api = NinjaAPI(parser=ORJSONParser())

class Payload(Schema):
    ints: str
    string: str
    f: float

@api.post('/json')
def operation(request, payload: Payload):
    return payload.dict()

过滤filtering

如果想通过许多不同的属性来过滤查询集，则将过滤器封装到 FilterSchema 类中是有意义的。 FilterSchema 是一个常规 Schema，因此它使用了 Pydantic 的所有必要功能，但它还添加了一些花哨的功能，可以轻松地将面向用户的过滤参数转换为数据库查询。

FilterSchema 子类

from ninja import FilterSchema, Field, Query
from typing import Optional
from django.core import serializers

class BookFilterSchema(FilterSchema):
    first_name: Optional[str]
    last_name: Optional[str]


@router.get("/books")
def list_books(request, filters: BookFilterSchema = Query(...)):
    books = Employee.objects.all()
    book = filters.filter(books)
    # QuerySet 需要序列号返回
    seria_books = serializers.serialize("json", book)

    return seria_books

Q表达式

@router.get("/q/books")
def list_books(request, filters: BookFilterSchema = Query(...)):

    # Never serve books from inactive publishers and authors
    q = Q(first_name__is_active=True) | Q(last_name__is_active=True)

    # But allow filtering the rest of the books
    q &= filters.get_filter_expression()
    return Employee.objects.filter(q)

自定义字段

当数据库查询比这更复杂时，你可以在字段定义中使用 “q “kwarg明确指定它们

class BookFilterSchema(FilterSchema):
    search: Optional[str] = Field(q=['name__icontains',
                                     'author__name__icontains',
                                     'publisher__name__icontains'])

组合表达式

字段级表达式使用 OR 运算符连接在一起。
字段本身使用 AND 运算符连接在一起。
可以在字段级和类级定义中使用 expression_connector 参数自定义此行为:

class BookFilterSchema(FilterSchema):
    active: Optional[bool] = Field(q=['is_active', 'publisher__is_active'],
                                   expression_connector='AND')
    name: Optional[str] = Field(q='name__icontains')

    class Config:
        expression_connector = 'OR'

表达式连接器可以采用“OR”、“AND”和“XOR”的值，但后者仅在 Django 4.1 中受支持。

无过滤

可以使 FilterSchema 将 None 视为应进行过滤的有效值。
这可以使用ignore_none kwarg在字段级别完成:

class BookFilterSchema(FilterSchema):
    name: Optional[str] = Field(q='name__icontains')
    tag: Optional[str] = Field(q='tag', ignore_none=False)

自定义表达式

有时可能想拥有复杂的过滤场景，而这些场景不能由单个字段注释来处理。对于这种情况，你可以把你的字段过滤逻辑作为一个自定义方法来实现。简单地定义一个名为filter_的方法，它接收一个过滤值并返回一个Q表达式:

class BookFilterSchema(FilterSchema):
    tag: Optional[str]
    popular: Optional[bool]

    def filter_popular(self, value: bool) -> Q:
        return Q(view_count__gt=1000) | Q(download_count__gt=100) if value else Q()

这样的字段方法优先于相应字段的Field()定义中的内容。
如果这还不够，可以在一个custom_expression方法中为整个FilterSet类实现你自己的自定义过滤逻辑：

class BookFilterSchema(FilterSchema):
    first_name: Optional[str]
    last_name: Optional[str]

    def custom_expression(self) -> Q:
        q = Q()
        if self.first_name:
            q &= Q(first_name__icontains=self.first_name)
        if self.last_name:
            q &= (
                Q(department_id__gt=10) |
                Q(cv='003644wOJAG')
            )
        return q

custom_expression方法优先于前面描述的任何其他定义，包括filter_方法。

处理响应

Response Schema

models.py中添加模型表字段

from django.contrib.auth.models import AbstractBaseUser
class User(AbstractBaseUser):
    username = models.CharField(max_length=50)
    password = models.CharField(max_length=30)

settings.py配置：

1
2
3

PASSWORD_HASHERS = (
    'django.contrib.auth.hashers.MD5PasswordHasher',
)

from ninja import FilterSchema, Field, Router, Query
from ninja import Schema
from system.models import *

api = Router()


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):
    print(data.username)
    print(data.password)
    user = User(username=data.username)
    user.set_password(data.password)
    user.save()

嵌套对象 - Nested objects

假设我们有一个带有用户外键的任务 Django 模型：

class Task(models.Model):
    title = models.CharField(max_length=200)
    is_completed = models.BooleanField(default=False)
    owner = models.ForeignKey("auth.User", null=True, blank=True, on_delete=models.CASCADE)

通常还需要返回一些嵌套/子对象的响应。

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.select_related("owner")
    return list(queryset)

别名 - Aliases

Ninja Schema 对象扩展了 Pydantic 的 Field(…, alias=””) 格式以处理点响应。

from ninja import Field, Schema


class TaskSchema(Schema):
    id: int
    title: str
    # The first Field param is the default, use ... for required fields.
    completed: bool = Field(..., alias="is_completed")
    owner_first_name: str = Field(None, alias="owner.first_name")

别名响应示例图

别名还支持 django 模板语法变量访问：

1
2
3

class TaskSchema(Schema):
    last_message: str = Field(None, alias="message_set.0.text")

解析器 - Resolvers

也可以通过基于字段名的解析方法创建计算字段。
该方法必须接受一个单一的参数，这将是模式所要解析的对象。
当作为一个标准方法创建解析器时，self使你能够访问模式中其他经过验证和格式化的属性。

class TaskSchema(Schema):
    id: int
    title: str
    is_completed: bool
    owner: Optional[str]
    lower_title: str
    email: str

    @staticmethod
    def resolve_owner(obj):
        if not obj.owner:
            return
        return f"{obj.owner.first_name} {obj.owner.last_name}"

    def resolve_lower_title(self, obj):
        return self.title.lower()

    def resolve_email(self, obj):
        return f"{obj.owner.email}"

解析器响应图

返回查询集

1
2
3

@api.get("/tasks/all", response=List[TaskSchema])
def tasks(request):
    return Task.objects.all()

如果您的操作是异步的，则此示例将不起作用，因为需要安全地调用 ORM 查询。

文件字段和图像字段

Django Ninja默认将文件和图片（用FileField或ImageField声明）转换为字符串URL。
models.py创建模型表：

1
2
3

class Picture(models.Model):
    title = models.CharField(max_length=100)
    image = models.ImageField(upload_to='images')

**api**示例代码：

from ninja import Schema, File
from ninja.files import UploadedFile
from system.models import *


class PictureSchema(Schema):
    title: str


class PictureRsp(Schema):
    title: str
    image: str


@api.post("/tasks/upload", response=PictureRsp)
def tasks(request, item: PictureSchema, img: UploadedFile = File(...)):
    payload_dict = item.dict()
    picture = Picture(**payload_dict)
    picture.image.save(img.name, img)
    return picture

文件字段和图像字段响应图

多重响应模式

有时你需要定义多个响应模式。例如，在认证的情况下，你可以返回：

200 successful -> token
401 -> Unauthorized
402 -> Payment required
etc..
可以向响应参数传递一个字典，其中：
key 是响应代码

value 是该代码的模式
另外，当你返回结果时，你还必须传递一个状态码来告诉Django Ninja应该使用哪种模式来进行验证和序列化。
如下：

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': 'Insufficient balance amount. Please proceed to a payment page.'}
    return 200, {'token': xxx, ...}

多个响应代码

为了避免消息模式重复，可以为一个模式使用多个响应代码：

...
from ninja.responses import codes_4xx


@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, ...}

Django Ninja自带以下HTTP代码：

from ninja.responses import codes_1xx
from ninja.responses import codes_2xx
from ninja.responses import codes_3xx
from ninja.responses import codes_4xx
from ninja.responses import codes_5xx

也可以使用frozenset创建你自己的范围：

1
2
3

my_codes = frozenset({416, 418, 425, 429, 451})

@api.post('/login', response={200: Token, my_codes: Message})

空响应

有的响应，例如204无内容，没有主体。为了表示响应的主体是空的，可以用None而不是Schema来标记响应参数：

1
2
3

@api.post("/no_content", response={204: None})
def no_content(request):
    return 204, None

自引用方案

有时需要创建一个引用自身或树结构对象的模式。
为此，需要：

在引号中设置模式类型

使用 update_forward_refs 方法应用自引用类型

class Organization(Schema):
    title: str
    part_of: 'Organization' = None     #!! note the type in quotes here !!


Organization.update_forward_refs()  # !!! this is important


@api.get('/organizations', response=List[Organization])
def list_organizations(request):
    ...

create_schema()自引用方案

为了能够从通过 create_schema() 生成的模式中使用 update_forward_refs() 方法，类的“名称”需要位于我们的命名空间中。在这种情况下，将 name 参数传递给 create_schema() 非常重要

UserSchema = create_schema(
    User,
    name='UserSchema',  # !!! this is important for update_forward_refs()  
    fields=['id', 'username']
    custom_fields=[
        ('manager', 'UserSchema', None),
    ]
)
UserSchema.update_forward_refs()

在视图之外序列化

你的对象的序列化可以通过使用模式对象上的.from_orm()方法直接在代码中完成。
创建如下模型：

1 2	class Person(models.Model): name = models.CharField(max_length=50)

如下schema：

1 2	class PersonSchema(Schema): name: str

可以使用模式上的 .from_orm() 方法执行直接序列化。拥有模式对象的实例后，.dict() 和 .json() 方法允许您获取字典输出和字符串 JSON 版本。

>>> person = Person.objects.get(id=1)
>>> data = PersonSchema.from_orm(person)
>>> data
PersonSchema(id=1, name='Mr. Smith')
>>> data.dict()
{'id':1, 'name':'Mr. Smith'}
>>> data.json()
'{"id":1, "name":"Mr. Smith"}'

多个项目：或查询集（或列表）:

1
2
3

>>> persons = Person.objects.all()
>>> data = [PersonSchema.from_orm(i).dict() for i in persons]
[{'id':1, 'name':'Mr. Smith'},{'id': 2, 'name': 'Mrs. Smith'}...]

Django HTTP responses

也可以返回常规的django http响应：

from django.http import HttpResponse
from django.shortcuts import redirect


@api.get("/http")
def result_django(request):
    return HttpResponse('some data')   # !!!!


@api.get("/something")
def some_redirect(request):
    return redirect("/some-path")  # !!!!

改变响应

有时您会想在响应提供之前更改响应，例如添加标头或更改 cookie。
为此，只需声明一个 HttpResponse 类型的函数参数：

from django.http import HttpRequest, HttpResponse

@api.get("/cookie/")
def feed_cookiemonster(request: HttpRequest, response: HttpResponse):
    # Set a cookie.
    response.set_cookie("cookie", "delicious")
    # Set a header.
    response["X-Cookiemonster"] = "blue"
    return {"cookiemonster_happy": True}

更改基本响应对象

可以通过重写 NinjaAPI.create_temporal_response 方法来更改此时间响应对象。

def create_temporal_response(self, request: HttpRequest) -> HttpResponse:
    response = super().create_temporal_response(request)
    # Do your magic here...
    return response

Django模型的模式

模式对于定义验证规则和响应非常有用，但有时您需要将数据库模型反映到模式中并保持更改同步。

ModelSchema

ModelSchema 是一个特殊的基类，可以从模型自动生成模式。
所需要做的就是在模式配置上设置 model 和 model_fields 属性：

from django.contrib.auth.models import User
from ninja import ModelSchema

class UserSchema(ModelSchema):
    class Config:
        model = User
        model_fields = ['id', 'username', 'first_name', 'last_name']

# Will create schema like this:
# 
# class UserSchema(Schema):
#     id: int
#     username: str
#     first_name: str
#     last_name: str

使用所有模型字段

要使用模型中的所有字段 - 可以将 __all__ 传递给 model_fields：

class UserSchema(ModelSchema):
    class Config:
        model = User
        model_fields = "__all__"

不建议全部使用。
这可能会导致意外的不需要的数据泄露（例如上例中的散列密码）。
一般建议 - 使用 model_fields 显式定义您希望在 API 中可见的字段列表。

排除模型字段

要使用除少数字段之外的所有字段，您可以使用 model_exclude 配置：

class UserSchema(ModelSchema):
    class Config:
        model = User
        model_exclude = ['password', 'last_login', 'user_permissions']

# Will create schema like this:
# 
# class UserSchema(Schema):
#     id: int
#     username: str
#     first_name: str
#     last_name: str
#     email: str
#     is_superuser: bool
#     ... and the rest

覆盖字段

要更改某些字段的默认注释或添加新字段，只需照常使用带注释的属性即可。

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']

字段可选

对于 PATCH API 操作，您通常需要将模式的所有字段设为可选。为此，可以使用配置 model_fields_optional:

class PatchGroupSchema(ModelSchema):
    class Config:
        model = Group
        model_fields = ['id', 'name', 'description'] # Note: all these fields are required on model level
        model_fields_optional = '__all__'

也可以定义多个可选的字段而不是全部：

1	model_fields_optional = ['name', 'description']

使用create_schema

在底层，ModelSchema 使用 create_schema 函数。这是一种更高级（且不太安全）的方法 - 请谨慎使用。
Django Ninja 带有一个辅助函数 create_schema：

def create_schema(
    model, # django model
    name = "", # name for the generated class, if empty model names is used
    depth = 0, # if > 0 schema will be also created for the nested ForeignKeys and Many2Many (with the provided depth of lookup)
    fields: list[str] = None, # if passed - ONLY these fields will added to schema
    exclude: list[str] = None, # if passed - these fields will be excluded from schema
    optional_fields: list[str] | str = None, # if passed - these fields will not be required on schema (use '__all__' to mark ALL fields required)
    custom_fields: list[tuple(str, Any, Any)] = None, # if passed - this will override default field types (or add new fields)
)

示例如下：

from django.contrib.auth.models import User
from ninja.orm import create_schema

UserSchema = create_schema(User)

# Will create schema like this:
# 
# class UserSchema(Schema):
#     id: int
#     username: str
#     first_name: str
#     last_name: str
#     password: str
#     last_login: datetime
#     is_superuser: bool
#     email: str
#     ... and the rest

默认情况下，create_schema 会构建包含所有模型字段的模式。这可能会导致意外的不需要的数据泄露（例如上例中的散列密码）。
始终使用字段或排除参数来显式定义属性列表。

使用字段 - fields

UserSchema = create_schema(User, fields=['id', 'username'])

# Will create schema like this:
# 
# class UserSchema(Schema):
#     id: int
#     username: str

使用排除 - exclude

UserSchema = create_schema(User, exclude=[
    'password', 'last_login', 'is_superuser', 'is_staff', 'groups', 'user_permissions']
)

# Will create schema without excluded fields:
# 
# class UserSchema(Schema):
#    id: int
#    username: str
#    first_name: str
#    last_name: str
#    email: str
#    is_active: bool
#    date_joined: datetime

使用深度 - depth

The depth argument allows you to introspect the Django model into the Related fields(ForeignKey, OneToOne, ManyToMany).

UserSchema = create_schema(User, depth=1, fields=['username', 'groups'])

# Will create the following schema:
#
# class UserSchema(Schema):
#    username: str
#    groups: List[Group]

此处的组变成了 List[Group] - Many2many 字段内省了 1 级并为组创建了模式：

class Group(Schema):
    id: int
    name: str
    permissions: List[int]

重写 Pydantic 配置

通过架构的 Pydantic Config 类，可以对 Django Ninja 架构进行许多自定义。

Django Ninja 使用 Pydantic 模型及其所有功能和优点。选择别名 Schema 是为了避免使用 Django 模型时代码混乱，因为 Pydantic 的模型类默认称为 Model，并且与 Django 的 Model 类冲突。

驼峰模式

一个有趣的 Config 属性是 alias_generator。在 Django Ninja 中使用 Pydantic 的示例可能类似于：

from ninja import Schema


def to_camel(string: str) -> str:
    return ''.join(word.capitalize() for word in string.split('_'))


class CamelModelSchema(Schema):
    str_field_name: str
    float_field_name: float

    class Config(Schema.Config):
        alias_generator = to_camel

当重写模式的 Config 时，需要从 Config 基类继承。

想要修改字段名称的输出（例如驼峰命名法）时 - 还需要设置allow_population_by_field_name和by_alias

from ninja import ModelSchema


def to_camel(string: str) -> str:
    return ''.join(word.capitalize() for word in string.split('_'))


class UserSchemas(ModelSchema):
    class Config:
        model = User
        model_fields = ["username"]
        alias_generator = to_camel
        allow_population_by_field_name = True  # !!!!!! <--------


@api.get("/users", response=list[UserSchemas], by_alias=True) # !!!!!! <-------- by_alias
def get_users(request):
    return User.objects.all()

驼峰模式示例输出图

Django 模型的自定义配置

使用 create_schema 时，生成的模式可用于构建另一个具有自定义配置的类，例如：

from django.contrib.auth.models import User
from ninja.orm import create_schema


BaseUserSchema = create_schema(User)


class UserSchema(BaseUserSchema):

    class Config(BaseUserSchema.Config):
        ...

Django Ninja 附带分页支持。这允许您将大型结果集拆分为单独的页面。

要将分页应用于函数 - 只需应用分页装饰器：

from ninja.pagination import paginate

@api.get('/users', response=List[UserSchema])
@paginate
def list_users(request):
    return User.objects.all()

可以使用 limit 和 offset GET 参数查询用户

/api/users?limit=10&offset=0

默认限制设置为 100（您可以使用 NINJA_PAGINATION_PER_PAGE 在 settings.py 中更改它）

内置分页类

LimitOffsetPagination (default)
这是默认的分页类（您可以使用 NINJA_PAGINATION_CLASS 类路径在 settings.py 中更改它）

from ninja.pagination import paginate, LimitOffsetPagination

@api.get('/users', response=List[UserSchema])
@paginate(LimitOffsetPagination)
def list_users(request):
    return User.objects.all()

该类有两个输入参数：

limit - 定义页面上的查询集数量（默认 = 100，在 NINJA_PAGINATION_PER_PAGE 中更改）
offset - 设置页面窗口偏移量（默认值：0，索引从 0 开始）

PageNumberPagination
该类有一个参数页，页码从 1 开始,默认每页输出 100 个查询集（可以使用 settings.py 进行更改）

from ninja.pagination import paginate, PageNumberPagination

@api.get('/users', response=List[UserSchemas])
@paginate(PageNumberPagination)
def list_users(request):
    return User.objects.all()

还可以为每个视图单独设置自定义 page_size 值：

1
2
3

@api.get("/users")
@paginate(PageNumberPagination, page_size=50)
def list_users(...

视图函数中访问分页器参数

如果您需要访问视图函数中用于分页的输入参数 - 使用 pass_parameter 参数
在这种情况下，输入数据将在 **kwargs 中可用：

@api.get("/someview")
@paginate(pass_parameter="pagination_info")
def someview(request, **kwargs):
    page = kwargs["pagination_info"].page
    return ...

创建自定义分页类

要创建自定义分页类，您应该子类化 ninja.pagination.PaginationBase 并覆盖输入和输出模式类以及 paginate_queryset(self, queryset, request, **params) 方法：

输入模式是一个模式类，它描述应该传递给分页器的参数（例如页码或限制/偏移值）。
输出模式描述页面输出的模式（例如计数/下一页/项目/等）。

paginate_queryset 方法会传递初始查询集，并应返回一个仅包含所请求页面中的数据的可迭代对象。该方法接受以下参数：

queryset：api 函数返回的查询集（或可迭代）
pagination - 分页器。输入参数（已解析和验证）

**params：kwargs 将包含装饰函数收到的所有参数
Example:

from ninja.pagination import paginate, PaginationBase
from ninja import Schema
from typing import List, Any


class CustomPagination(PaginationBase):
    # only `skip` param, defaults to 5 per page
    class Input(Schema):
        skip: int


    class Output(Schema):
        items: List[Any] # `items` is a default attribute
        total: int
        per_page: int

    def paginate_queryset(self, queryset, pagination: Input, **params):
        skip = pagination.skip
        return {
            'items': queryset[skip : skip + 5],
            'total': queryset.count(),
            'per_page': 5,
        }


@api.get('/users', response=List[UserSchemas])
@paginate(CustomPagination)
def list_users(request):
    return User.objects.all()

输出属性 - Output attribute

默认情况下，页面项目放置在“items”属性中。要覆盖此行为，请使用 items_attribute：

class CustomPagination(PaginationBase):
    ...
    class Output(Schema):
        results: List[Any]
        total: int
        per_page: int

    items_attribute: str = "results"

一次将分页应用于多个操作

通常情况下，需要向返回查询集或列表的所有视图添加分页

可以使用内置路由器类（RouterPaginated）自动将分页注入到定义response=List[SomeSchema]的所有操作中：

from ninja.pagination import RouterPaginated
from typing import List

router = RouterPaginated()

class UserSchemas(Schema):
    username: str


@api.get('/routers/users', response=List[UserSchemas])
def list_users(request):
    return User.objects.all()

要将分页应用于主 api 实例，请使用 default_router 参数：

1
2
3

api = NinjaAPI(default_router=RouterPaginated())

@api.get(...

响应渲染器 - Response renderers

REST API 最常见的响应类型通常是 JSON。 Django Ninja 还支持定义您自己的自定义渲染器，这使您可以灵活地设计自己的媒体类型。

创建渲染器

要创建自己的渲染器，需要继承ninja.renderers.BaseRenderer并重写render方法。然后，您可以将类的实例作为渲染器参数传递给 NinjaAPI：

渲染方法采用以下参数：

request -> HttpRequest 对象
data -> 需要序列化的对象
response_status as an int -> 将返回给客户端的 HTTP 状态代码

还需要在类上定义 media_type 属性来设置响应的内容类型标头。

ORJSON 渲染器

orjson 是一个快速、准确的 Python JSON 库。它被评为最快的 JSON Python 库，并且比标准 json 库或其他第三方库更准确。它还本机序列化数据类、日期时间、numpy 和 UUID 实例。

下面是一个使用 orjson 的示例渲染器类：

import orjson
from ninja import NinjaAPI
from ninja.renderers import BaseRenderer


class ORJSONRenderer(BaseRenderer):
    media_type = "application/json"

    def render(self, request, data, *, response_status):
        return orjson.dumps(data)

api = NinjaAPI(renderer=ORJSONRenderer())

XML 渲染器

这是创建将所有响应输出为 XML 的渲染器的方法：

from io import StringIO
from django.utils.encoding import force_str
from django.utils.xmlutils import SimplerXMLGenerator
from ninja import NinjaAPI
from ninja.renderers import BaseRenderer


class XMLRenderer(BaseRenderer):
    media_type = "text/xml"

    def render(self, request, data, *, response_status):
        stream = StringIO()
        xml = SimplerXMLGenerator(stream, "utf-8")
        xml.startDocument()
        xml.startElement("data", {})
        self._to_xml(xml, data)
        xml.endElement("data")
        xml.endDocument()
        return stream.getvalue()

    def _to_xml(self, xml, data):
        if isinstance(data, (list, tuple)):
            for item in data:
                xml.startElement("item", {})
                self._to_xml(xml, item)
                xml.endElement("item")

        elif isinstance(data, dict):
            for key, value in data.items():
                xml.startElement(key, {})
                self._to_xml(xml, value)
                xml.endElement(key)

        elif data is None:
            # Don't output any value
            pass

        else:
            xml.characters(force_str(data))


api = NinjaAPI(renderer=XMLRenderer())

路由器

多应用路由

当有多个应用时，在每个应用中的创建一个api.py模块(或者直接在views.py模块)中写各自的路由。

from datetime import date
from ninja import Router, Schema
from system.models import *


router = Router()

class EmployeeIn(Schema):
    first_name: str
    last_name: str
    department_id: int = None
    birthdate: date = None

@router.post("/employees")
def create_employee(request, payload: EmployeeIn):
    employee = Employee.objects.create(**payload.dict())
    return {"id": employee.id}

路由器结构图

路由器认证

使用 auth 参数将验证器应用于路由器声明的所有操作：

1	api.add_router("/events/", events_router, auth=BasicAuth())

或使用路由器构造函数:

1	router = Router(auth=BasicAuth())

路由器标签

可以使用标签参数将标签应用于路由器声明的所有操作：

1	api.add_router("/events/", events_router, tags=["events"])

或者使用路由器构造函数：

1	router = Router(tags=["events"])

嵌套路由器

有时您需要进一步拆分逻辑。 Django Ninja 可以将一个路由器多次包含到另一个路由器中，最后将顶级路由器包含到主 api 实例中。

from django.contrib import admin
from django.urls import path
from ninja import NinjaAPI, Router

api = NinjaAPI()

first_router = Router()
second_router = Router()
third_router = Router()


@api.get("/add")
def add(request, a: int, b: int):
    return {"result": a + b}


@first_router.get("/add")
def add(request, a: int, b: int):
    return {"result": a + b}


@second_router.get("/add")
def add(request, a: int, b: int):
    return {"result": a + b}


@third_router.get("/add")
def add(request, a: int, b: int):
    return {"result": a + b}


second_router.add_router("l3", third_router)
first_router.add_router("l2", second_router)
api.add_router("l1", first_router)

urlpatterns = [
    path("admin/", admin.site.urls),
    path("api/", api.urls),
]

然后就有如下路由路径：

/api/add
/api/l1/add
/api/l1/l2/add
/api/l1/l2/l3/add

Authentication

Django Ninja 提供了多种工具来帮助您以标准方式轻松、快速地处理身份验证和授权，而无需学习所有安全规范。
核心概念是，当你描述一个API操作时，你可以定义一个认证对象。

from ninja import NinjaAPI
from ninja.security import django_auth

api = NinjaAPI(csrf=True)


@api.get("/pets", auth=django_auth)
def pets(request):
    return f"Authenticated user {request.auth}"

在此示例中，客户端只有使用 Django 会话身份验证（默认基于 cookie）才能调用 pets 方法，否则将返回 HTTP-401 错误。

自动OpenAPI模式

如下示例，客户端为了进行身份验证，需要传递标头：

from ninja.security import HttpBearer


class AuthBearer(HttpBearer):
    def authenticate(self, request, token):
        if token == "supersecret":
            return token


@api.get("/bearer", auth=AuthBearer())
def bearer(request):
    return {"token": 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())

# 免认证
@api.post("/token", auth=None)  # < overriding global auth
def get_token(request, username: str = Form(...), password: str = Form(...)):
    if username == "admin" and password == "giraffethinnknslong":
        return {"token": "supersecret"}

可用的身份验证选项

自定义功能

“auth=”参数接受任何 Callable 对象。仅当可调用对象返回可转换为布尔值 True 的值时，NinjaAPI 才会通过身份验证。该返回值将分配给 request.auth 属性。

def ip_whitelist(request):
    if request.META["REMOTE_ADDR"] == "8.8.8.8":
        return "8.8.8.8"


@api.get("/ipwhiltelist", auth=ip_whitelist)
def ipwhiltelist(request):
    return f"Authenticated client, IP = {request.auth}"

api key

某些 API 使用 API 密钥进行授权。 API 密钥是客户端在进行 API 调用时提供的用于识别自身身份的令牌。密钥可以在查询字符串中发送：

GET /something?api_key=abcdef12345

或作为一个请求头：

GET /something HTTP/1.1
X-API-Key: abcdef12345

或作为 cookie：

GET /something HTTP/1.1
Cookie: X-API-KEY=abcdef12345

Django Ninja 附带了内置类来帮助处理这些情况。

in Query

# models.py
class Client(models.Model):
    key = models.CharField(max_length=100)


from ninja.security import APIKeyQuery
from system.models import Client


class ApiKey(APIKeyQuery):
    param_name = "api_key"

    def authenticate(self, request, key):
        try:
            return Client.objects.get(key=key)
        except Client.DoesNotExist:
            pass


api_key = ApiKey()


@router.get("/apikey", auth=api_key)
def apikey(request):
    assert isinstance(request.auth, Client)
    return f"Hello {request.auth}"

在此示例中，我们从 GET[‘api_key’] 获取一个令牌，并在数据库中找到与该密钥对应的客户端。 Client 实例将被设置为 request.auth 属性。

param_name 是要检查的 GET 参数的名称。如果未设置，将使用默认的“key”。

in Header

from ninja.security import APIKeyHeader


class ApiKey(APIKeyHeader):
    param_name = "X-API-Key"

    def authenticate(self, request, key):
        if key == "supersecret":
            return key


header_key = ApiKey()


@api.get("/headerkey", auth=header_key)
def apikey(request):
    return f"Token = {request.auth}"

from ninja.security import APIKeyCookie


class CookieKey(APIKeyCookie):
    def authenticate(self, request, key):
        if key == "supersecret":
            return key


cookie_key = CookieKey()


@api.get("/cookiekey", auth=cookie_key)
def apikey(request):
    return f"Token = {request.auth}"

HTTP Bearer

from ninja.security import HttpBearer


class AuthBearer(HttpBearer):
    def authenticate(self, request, token):
        if token == "supersecret":
            return token


@api.get("/bearer", auth=AuthBearer())
def bearer(request):
    return {"token": request.auth}

HTTP Basic Auth

from ninja.security import HttpBasicAuth


class BasicAuth(HttpBasicAuth):
    def authenticate(self, request, username, password):
        if username == "admin" and password == "secret":
            return username


@api.get("/basic", auth=BasicAuth())
def basic(request):
    return {"httpuser": request.auth}

多个验证器

auth 参数还允许您传递多个身份验证器：

from ninja.security import APIKeyQuery, APIKeyHeader


class AuthCheck:
    def authenticate(self, request, key):
        if key == "supersecret":
            return key


class QueryKey(AuthCheck, APIKeyQuery):
    pass


class HeaderKey(AuthCheck, APIKeyHeader):
    pass


@api.get("/multiple", auth=[QueryKey(), HeaderKey()])
def multiple(request):
    return f"Token = {request.auth}"

在这种情况下，Django Ninja 将首先检查 API 密钥 GET，如果未设置或无效，将检查标头密钥。如果两者都无效，则会向响应引发身份验证错误。

路由器认证

在 Router 上使用 auth 参数将验证器应用于其中声明的所有操作：

api.add_router(“/events/“, events_router, auth=BasicAuth())

或使用路由器构造函数:

router = Router(auth=BasicAuth())

自定义处理

引发具有异常处理程序的异常将从该处理程序返回响应，其方式与操作相同：

from ninja import NinjaAPI
from ninja.security import HttpBearer

api = NinjaAPI()

class InvalidToken(Exception):
    pass

@api.exception_handler(InvalidToken)
def on_invalid_token(request, exc):
    return api.create_response(request, {"detail": "Invalid token supplied"}, status=401)

class AuthBearer(HttpBearer):
    def authenticate(self, request, token):
        if token == "supersecret":
            return token
        raise InvalidToken


@api.get("/bearer", auth=AuthBearer())
def bearer(request):
    return {"token": request.auth}

异步认证

Django Ninja 对异步身份验证具有基本支持。虽然默认的身份验证类不兼容异步，但您仍然可以定义自定义异步身份验证可调用对象并使用 auth.

async def async_auth(request):
    ...


@api.get("/pets", auth=async_auth)
def pets(request):
    ...

异常处理

Django Ninja 允许安装自定义异常处理程序来处理发生错误或已处理异常时返回响应的方式。

自定义异常处理

如果你在写的 API 依赖于某些外部服务，而这些外部服务被设计为在某些时刻不可用。您可以处理错误并向客户端提供一些友好的响应（稍后返回），而不是在异常时抛出默认的 500 错误。
为此，可以：

创建一些异常（或使用现有的异常）
使用 api.exception_handler 装饰器

import random
from ninja import NinjaAPI

api = NinjaAPI()

class ServiceUnavailableError(Exception):
    pass


# initializing handler

@api.exception_handler(ServiceUnavailableError)
def service_unavailable(request, exc):
    return api.create_response(
        request,
        {"message": "Please retry later"},
        status=503,
    )


# some logic that throws exception

@api.get("/service")
def some_operation(request):
    if random.choice([True, False]):
        raise ServiceUnavailableError()
    return {"message": "Hello"}

异常处理函数有 2 个参数：

请求 - Django http 请求
exc - 实际异常

函数必须返回http响应。

覆盖默认的异常处理

默认情况下，Django Ninja 初始化了以下异常处理程序：
ninja.errors.AuthenticationError
身份验证数据无效时引发

ninja.errors.ValidationError
当请求数据未验证时引发

ninja.errors.HttpError
用于从代码的任何位置抛出带有状态代码的http错误

django.http.Http404
Django 默认的 404 异常（可以使用 get_object_or_404 返回）

Exception
应用程序的任何其他未处理的异常。

默认行为:

if settings.DEBUG 为 True - 以纯文本形式返回回溯（在控制台或 swagger UI 中调试时很有用）
else - 使用默认的 django 异常处理程序机制（错误日志记录、发送电子邮件至 ADMINS）

如果需要更改验证错误的默认输出 - 覆盖 ValidationError 异常处理程序：

from ninja.errors import ValidationError
...

@api.exception_handler(ValidationError)
def validation_errors(request, exc):
    return HttpResponse("Invalid input", status=422)

抛出异常的 HTTP 响应

作为自定义异常和为其编写处理程序的替代方案 - 您也可以抛出 http 异常，这将导致返回带有所需代码的 http 响应

from ninja.errors import HttpError

@api.get("/some/resource")
def some_operation(request):
    if True:
        raise HttpError(503, "Service Unavailable. Please retry later.")

操作参数

OpenAPI模式相关

tags
您可以使用标签参数 (list[str]) 对 API 操作进行分组。

1
2
3

@api.post("/orders/", tags=["orders"])
def create_order(request, order: Order):
    return {"success": True}

openapi_extra

您可以为特定端点自定义 OpenAPI 架构（详细信息 OpenAPI 自定义选项）

# You can set requestBody from openapi_extra
@api.get(
    "/tasks",
    openapi_extra={
        "requestBody": {
            "content": {
                "application/json": {
                    "schema": {
                        "required": ["email"],
                        "type": "object",
                        "properties": {
                            "name": {"type": "string"},
                            "phone": {"type": "number"},
                            "email": {"type": "string"},
                        },
                    }
                }
            },
            "required": True,
        }
    },
)
def some_operation(request):
    pass

# You can add additional responses to the automatically generated schema
@api.post(
    "/tasks",
    openapi_extra={
        "responses": {
            "400": {
                "description": "Error Response",
            },
            "404": {
                "description": "Not Found Response",
            },
        },
    },
)
def some_operation_2(request):
    pass

响应输出选项

by_alias
字段别名是否应用作响应中的键（默认为 False）。

exclude_unset
是否应从响应中排除创建架构时未设置且具有默认值的字段（默认为 False）。

exclude_defaults
是否应从响应中排除等于默认值（无论是否设置）的字段（默认为 False）。

exclude_none
是否应从响应中排除等于 None 的字段（默认为 False）。

url_name

允许您设置 api 端点 url 名称（使用 django 路径的命名）

@api.post("/tasks", url_name='tasks')
def some_operation(request):
    pass

# then you can get the url with

reverse('api-1.0.0:tasks')

指定服务器

如果要为 OpenAPI 规范指定单个或多个服务器，可以在初始化 NinjaAPI 实例时使用服务器：

from ninja import NinjaAPI

api = NinjaAPI(
        servers=[
            {"url": "https://stag.example.com", "description": "Staging env"},
            {"url": "https://prod.example.com", "description": "Production env"},
        ]
)

这将允许在使用交互式 OpenAPI 文档时在环境之间切换：

1	![交互式文档](https://i0.wp.com/cdn.jsdelivr.net/gh/lenkyes/GitImgStore/uploads/2023/07/20230717200231.png)

简介

启动项目

安装

创建项目

基础配置

启动项目

打开交互式文档

请求数据

路径参数

查询参数

分页查询

字符串匹配

使用Schema

请求体

使用类模型schema的方式：

请求体+路径参数

请求体+路径参数+查询参数

Form表单

Form表单数据作为参数

使用Schema

表单+路径参数+查询参数

可选表单空值设置

文件上传

单文件上传

多文件上传

上传带有额外字段的文件

请求解析

YAML 解析器

ORJSON 解析器

过滤filtering

FilterSchema 子类

Q表达式

自定义字段

组合表达式

无过滤

自定义表达式

处理响应

Response Schema

嵌套对象 - Nested objects

别名 - Aliases

解析器 - Resolvers

返回查询集

文件字段和图像字段

多重响应模式

多个响应代码

空响应

自引用方案

create_schema()自引用方案

在视图之外序列化

Django HTTP responses

改变响应

更改基本响应对象

Django模型的模式

ModelSchema

使用所有模型字段

排除模型字段

覆盖字段

字段可选

使用create_schema

使用字段 - fields

使用排除 - exclude

使用深度 - depth

重写 Pydantic 配置

驼峰模式

Django 模型的自定义配置

分页 - Pagination

内置分页类

视图函数中访问分页器参数

创建自定义分页类

输出属性 - Output attribute

一次将分页应用于多个操作

响应渲染器 - Response renderers

创建渲染器

ORJSON 渲染器

XML 渲染器

路由器

多应用路由

路由器认证

路由器标签

嵌套路由器