Django Ninja简单教程

Django Ninja

文章目录

Django Ninja
- 简介
- - 特点：
  - 交换式文档
- 教程
- - 搭建项目
  - 路由器
  - - 多应用路由
    - 路由器认证
    - 路由器标签
    - 路由器嵌套
  - 请求数据
  - - 路径参数
    - 请求参数
    - 请求体
    - 路径参数、查询参数和请求体
    - Form表单
    - 文件上传
  - Schema
  - - 请求相关Schema（参考请求数据中）
    - 响应体Schema
    - model的Schema
    - 其他
  - 认证
  - - 使用自带认证
    - 使用自定义认证
  - 操作参数
  - - 标签
    - 操作：摘要
    - 操作：说明
    - OpenAPI操作ID
    - 操作：已弃用
    - 输出响应选项
    - 从架构中包含/排除操作（文档）
    - 网址名称
  - 版本控制
  - - 不同的API版本号
    - 不同的业务逻辑
  - 请求解析器
  - - YAML解析器
    - ORJSON 解析器
  - 响应渲染器
  - - 创建渲染器
    - ORJSON渲染实例
    - XML渲染实例
  - 错误处理
  - - 自定义异常处理
    - 覆盖默认异常处理程序
    - 抛出异常的HTTP响应
  - CSRF
  - - 注意：将API与基于cookie的身份验证一起使用是不安全！
- 异步支持
- - 简单介绍
  - 快速示例
  - 混合同步和异步操作
  - 弹性搜索示例
  - 使用ORM

简介特点：

简单：旨在易于使用和直观；
快速执行：由于**Pydantic和异步支持，**性能非常高；
快速编码：类型提示和自动文档让您只关注业务逻辑；
基于标准：基于 API 的开放标准：OpenAPI（以前称为 Swagger）和JSON Schema；
Django 友好：（显然）与 Django 核心和 ORM 有很好的集成；
生产就绪：多家公司在实时项目中使用。

交换式文档

启动项目，然后访问http://127.0.0.1:8000/api/docs；
看到自动的交互式API文档（由OpenAPI/Swagger UI提供）。

教程搭建项目

安装
- ```
pip install django-ninja
```

创建django项目

python django-admin startproject myproject

单应用项目
- 应用创建应用urls.py同目录下的api.py（也可以直接在view.py中实现）
  - ```
  from ninja import NinjaAPI
  
  api = NinjiaAPI()
  
  @api.get("/hello")
  def hello(request):
      dosomething
```
- 在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(path)；如果是多个方法一个处理函数，则使用@api.api_operation(method, path)。其中，method用列表表示。

路由器

多应用路由

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

from ninja import Router
from .models import xxx

router = Router()

@router.get("/")
def list_events(request):
    return [
        {"id": elem.id, "title": elem.title}
        for elem in xxx.objects.all()
    ]

在项目文件夹urls.py中实现多个应用的router注册

from ninja import NinjaAPI 
from xxx.api import router as xxx_router
from yyy.api import router as yyy_router

api = NinjaAPI()

api.add_router("/xxx/", xxx_router)
api.add_router("/yyy/", yyy_router)

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

路由器认证

api.add_router("/xxx/", xxx_router, auth=BasicAuth())
# or we can write as this
# router = Router(auth=BasicAuth())

路由器标签

可以使用tags参数将标签应用于路由器声明的操作。

api.add_router("/xxx/", xxx_router, tags=["xxx"])
# or we can write as this
# router = Router(tags=["xxx"])

路由器嵌套

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

请求数据

路径参数

所有的路径参数都会按照给定的类型自动转化，如果转化失败，则报错。

常规python格式化字符串形式

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

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

django路径参数转换器

@api.get("/items/{int:item_id}")
def read_item(request, item_id):
    return {"item_id": item_id}

注：{int:item_id}之间不允许有空格，有空格会报错。

使用Schema

import datetime
from ninja import Schema, Path

class PathSchema(Schema):
    year: int
    month: int
    day: int
        
    def value(self):
        return datetime.date(self.year, self.month, self.day)

    
@api.get("/events/{year}/{month}/{day}")
def events(request, date: PathSchema = Path(...))

注：Path()函数用来标记date参数是路径参数。

请求参数

请求参数分为两种：位置参数和可选参数。

位置参数。不给定参数类型，则默认为字符串类型。

@api.get("/weapons")
def list_weapons(request, limit, offset):
    # type(limit) == str
    # type(offset) == str

可选参数。通过给定参数默认值实现。

@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

注意：如果使用None作为默认值，则该参数可传可不传。

路径参数、查询参数和请求体
- ```
from ninja import Schema

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

@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，则该参数为请求体。

Form表单

Form数据作为参数

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]
```
- 上传的文件属性和方法和django中相同，主要包括以下：
  - read()
    - 从文件中读取整个上传的文件。文件太大会报错。
  - multiple_chunks(chunk_size=None)
    - 如果上传的文件足够大，需要分块读取，返回True。默认情况下是大于2.5M的文件。
  - chunks(chunk_size=None)
    - 一个生成器，返回文件的块。
  - name
    - 上传文件的文件名
  - size
    - 上传文件的大小，以字节为单位
  - content_type
    - 与文件一起上传的内容类型头
  - content_type_extra
    - 包含传递给content-type头的额外参数的字典。
  - charset
    - 对于text/*内容类型，浏览器提供的字符集。

Schema

一般在应用中创建一个schema.py存储。
请求相关Schema（参考请求数据中）

响应体Schema

返回体为简单对象

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):
    di: 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)
)

使用model参数

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

UserSchema = create_schema(User)

注: 仅仅使用model参数会将所有的User信息返回，容易暴露敏感数据。

使用fields参数

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

使用exclude参数

UserSchema = create_schema(User, exclude=["password", "last_login", ...])

使用depth参数

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

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

自定义schema
- 参考官方文档：
  - https://django-ninja.rest-framework.com/tutorial/config-pydantic/

认证

一般建议把认证相关可调用对象放在util包中。

使用自带认证

使用django验证

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)

访问"/pets"路由时，会使用Django会话身份验证（默认是基于cookie），验证通过调用对应视图函数，否则返回HTTP-401错误。

全局认证

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

使用自定义认证

自定义功能
- "auth="参数接受任何Callable对象。仅当可调用对象返回可转化为布尔值True的值时，NinjaAPI才通过身份验证。此返回值将分配给属性request.auth。

请求参数中带有api_key验证信息

# GET /something?api_key=abcdefg12345
from ninja.security import APIKeyQuery
from someapp.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.get("/apikey", auth=ApiKey())
def apikey(request):
    assert isinstance(request.auth, Client)
    return "Hello {}".format(request.auth)

param_name是江北检查的GET参数的部分。如果未设置，将使用默认值"key"。

请求头中带有X-API-KEY验证信息

# GET /something HTTP/1.1
# X-API-Key: abcdef12345
from ninja.security import APIKeyHeader

class ApiKey(APIKeyHeader):
    param_name = "X-API-Key"
    def authenticate(self, request, key):
        if key == "supersecret":
            return key
        
@api.get("/headerKey", auth=ApiKey())
def apikey(request):
    return "Token = {}".format(request.auth)

cookie中带有验证

# GET /something HTTP/1.1
# cookie: X-API-KEY=abcdef12345
from ninja.security import APIKeycookie

class cookieKey(APIKeycookie):
    def authenticate(self, request, key):
        if key = "supersecret":
            return key
        
@api.get("/cookiekey", auth=cookieKey())
def apikey(request):
    return "Token = {}".format(request.auth)

HTTP JWT 验证

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基本身份验证

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}

多个验证器

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 "Token = {}".format(request.auth)

改变出现错误时返回值（自定义异常）

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

操作参数

标签

OpenAPI上分组依据，默认按router分组。

使用tags参数（list[str]）对API操作进行分组

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

路由器标签

api.add_router("/xxx/", xxx_router, tags=['xxx'])
# or we can write like this
# router = Router(tags=["xxx"])

操作：摘要
- OpenAPI上操作的可读名称，默认为视图函数名称大写生成。
- 使用summary参数进行修改。
  - ```
  @api.get("/hello", summary="Say Hello")
  def hello(request, name: str):
      return {"hello": name}
```
操作：说明
- OpenAPI上显示操作的的说明信息。
  - ```
  @api.post("/orders", description="Create an order and updates stock")
  def create_order(request, order: Order):
      return {"success": True}
```
- 注：当需要提供很长的多行描述时，可以使用 Pythondocstrings进行函数定义：

OpenAPI操作ID

OpenAPIoperationId是一个可选的唯一字符串，用于标识操作。如果提供，这些 ID 在您的 API 中描述的所有操作中必须是唯一的。
默认情况下，Django Ninja将其设置为module name+ function name。

每个操作单独设置。

@api.post("/tasks", operation_id="create_task")
def new_task(request):
    ...

覆盖全局行为。

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

操作：已弃用

将操作标记为已弃用。

使用deprecated参数。

@api.post("/make-order", deprecated=True)
def xxx_old_method(request, order: Order):
    return {"success": True}

输出响应选项
- by_alias：使用应将字段别名用作响应中的键（默认为False）。
- excluede_unset：是否应将从响应中排除在创建构架时未设置且具有默认值的字段（默认为False）。
- exclude_defaults：是否应从响应中排除等于其默认值（无论是否设置）的字段（默认为False）。
- exclude_none：是否从响应中排除等于None的字段（默认为False）
从架构中包含/排除操作（文档）
- 从OpenAPI模式中排除某些操作。
- 使用include_in_schema参数。
  - ```
  @api.poat("/hidden", include_in_schema=False)
  def xxx_hiden_operation(request):
      pass
```

网址名称

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

@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),
]

不同的业务逻辑

...
api = NinjaAPI(auth=token_auth, urls_namespace="public_api")
...
api_private = NinjaAPI(auth=session_auth, urls_namespace="pricate_api")
...
urlpatterns = [
    ...
    path("api/", api.urls),
    path("internal-api/", api_private.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())

render函数参数如下:
- request: HttpRequest对象
- data：需要序列化的对象
- response_status（int）：将返回给客户端的HTTP状态码

ORJSON渲染实例

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

import orjson
from ninja import NinjaAPI
from ninja.renderers import baseRender

class ORJSONRenderer(baseRender):
    media_type = "application/json"
    def render(self, request, data, *, status_code):
        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 SimpleXMLGenerator
from ninja import NinjaAPI
from ninja.renderers import baseRenderer

class XMLRenderer(baserRenderer):
    media_type = "text/xml"
    def render(self, request, data, *, status_code):
        stream = StringIO()
        xml = SimpleXMLGenerator(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:
            pass
        else:
            xml.characters(force_str(data))

api = NinjaAPI(renderer=XMLRenderer())

错误处理

自定义异常处理

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

api = NinjaAPI()

class ServiceUnavailableError(Exception):
    pass

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

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

异常处理函数有两个参数：
- 请求：Django http请求
- exc: 实际异常
函数必须返回http响应

覆盖默认异常处理程序
- 默认初始化异常
  - ninja.errors.ValidationError: 当请求数据未验证时引发
  - ninja.errors.HttpError：用于从代码的任何位置抛出带有状态代码的http错误
  - django.http.Http404：Django的默认404异常
  - Exception: 应用程序的任何其他未处理的异常
- 覆盖默认处理程序
  - ```
  from ninja.errors import ValidationError
  ...
  @api.exception_handler(ValidationError)
  def validation_errors(request, exc):
      return HttpResponse("Invalid input", status_code=422)
```

抛出异常的HTTP响应

from ninja.errors import HttpError

@api.get("/xxx/resource")
def xxx_operation(request):
    if True:
        raise HttpError(503, "Service Unavailable. please retry later.")

CSRF

默认情况下，Django Ninja对所有操作都关闭了CSRF 。要打开它，您需要使用csrfNinjaAPI 类的参数：
- ```
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)

异步支持简单介绍

从3.1开始，Django开始支持异步视图。
异步视图在以下方面更有效的工作：
- 通过网络调用外部API；
- 执行/等待数据库查询；
- 从/向磁盘驱动器读取/写入

快速示例

import time

@api.get("/say-after")
async def say_after(request, delay: int, word: str):
    await asyncio.sleep(delay)
    return {"saying": word}

注意：要运行此代码，必须使用Uvicorn或Daphne这样的ASGI服务器。

使用Uvicorn

安装
- ```
pip install uvicorn
```

启动服务器

uvicorn your_project.asgi.application --reload

混合同步和异步操作

项目中可以同时使用同步和异步操作，Django Ninja会指自动路由。

@api.get("/sya-sync")
def say_after_sync(request, delay: int, word: str):
    time.sleep(delay)
    return {"saying": word}

@api.get("/say_async")
async def say_after_async(request, delay: int, word: str):
    await asyncio.sleep(delay)
    return {"saying": word}

弹性搜索示例

from ninja import NinjaAPI
from elasticsearch import AsyncElasticsearch

api = NinjaAPI()
es = AsyncElasticsearch()

@api.get("/search")
async def search(request, q: str):
    resp = await es.search(
        index = "document",
        body={"query": {"query_string": {"query": q}}},
        size=20   
    )
    return resp["hits"]

使用ORM

目前，（2020 年 7 月）Django 的某些关键部分无法在异步环境中安全运行，因为它们具有无法感知协程的全局状态。Django 的这些部分被归类为“async-unsafe”，并且在异步环境中不会被执行。ORM是主要示例，但还有其他部分也以这种方式受到保护。

如果直接用异步操作ORM，则会报错。

# 会报错
@api.get("/blog/{post_id}")
async def search(request, post_id: int):
    blog = Blog.objects.get(pk=post_id)
    ...

使用装饰器，将同步转化为异步

from asgiref.sync import sync_to_async

@sync_to_async
def get_blog(post_id):
    return Blog.objects.get(pk=post_id)

@api.get("/blog/{post_id}")
async def search(request, post_id: int):
    blog = await get_blog(post_id)
    ...

不适用装饰器

@api.get("/blog/{post_id}")
async def search(request, post_id: int):
    blog = await sync_to_async(Blog.objects.get)(pk=post_id)

不会理解执行

all_blogs = await sync_to_async(Blog.objects.all)()

会立即执行

all_blogs = await sync_to_async(list)(Blog.objects.all())

参考文献：

[1]

Django Ninja简单教程

Python相关栏目本月热门文章