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 NinjaAPIapi = NinjiaAPI()@api.get("/hello")
def hello(request):dosomething

    • 在url.py配置url路由

      • from django.contrib import admin
from django.urls import path
from .api import apiurlpatterns = [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 xxxrouter = 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_routerapi = 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, RouterAPI = 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, Pathclass PathSchema(Schema):year: intmonth: intday: intdef 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 Listfrom pydantic import Fieldfrom ninja import Query, Schemaclass Filters(Schema):limit: int = 100offset: int = Nonequery: str = Nonecategory__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 Schemaclass Item(Schema):name: strdescription: str = Noneprice: floatquantity: int@api.post("/items")
def create(request, item: Item):return item
        • 注意：如果使用None作为默认值，则该参数可传可不传。

  • 路径参数、查询参数和请求体

    • from ninja import Schemaclass Item(Schema):name: strdescription: str = Noneprice: floatquantity: 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, Formclass Item(Schema):name: strdescription: str = Noneprice: floatquantity: int@api.post("/item")
def create(request, item: Item = Form(...)):return item

    • 路径参数，请求参数和表单

      • from ninja import Schema, Formclass Item(Schema):name: strdescription: str = Noneprice: floatquantity: 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, TypeVarPydanticField = TypeVar("PydanticField")class EmptyStrToDefault(Generic[PydanticField]):@classmethoddef __get_validators__(cls):yield cls.validate@classmethoddef validate(cls, value: PydanticField, field: ModelField) -> PydanticField:if value == "":return field.defaultreturn valueclass Item(Schema):name: strdescription: str = Noneprice: EmptyStrToDefault[float] = 0.0quantity: EmptyStrToDefault[int] = 0in_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 Schemaclass UserIn(Schema):username: strpassword: strclass UserOut(Schema):id: intusername: 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 modelsclass 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 Schemaclass UserSchema(Schema):id: intfirst_name: strlast_name: strclass TaskSchema(Schema):di: inttitle: stris_completed: boolowner: 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: strimage: str

    • 返回状态码和数据

      • from ninja import Schemaclass Token(Schema):token: strexpires: dateclass 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: strpart_of: 'Organization' = NoneOrganization.update_forward_refs()  # this is important

  • model的Schema

    • 选择包含字段

      • from django.contrib.auth.models import User
from ninjia import ModelSchemaclass UserSchema(ModelSchema):class Config:model = Usermodel_fields = ["id", "username", "first_name", "last_name"]

    • 选择不包含字段

      • from django.contrib.auth.models import User
from ninja import ModelSchemaclass UserSchema(ModelSchema):class Config:model = Usermodel_exclude = ["password", "last_login", "user_permissions"]

    • 覆盖字段(修改某些字段注释或者添加新字段)

      • class GroupSchema(ModelSchema):class Config:model = Groupmodel_fields = ["id", "name"]class UserSchema(ModelSchema):groups: List[GroupSchema] = []class Config:model = Usermodel_fields = ["id", "username", "first_name", "last_name"]

  • 其他

    • 使用create_schema

      • 语法

        • def create_schema(model, # django modelname = "",  # name for the genarated class ,if empty model name is useddepth = 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_schemaUserSchema = 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_authapi = 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 HttpBearerclass GlobalAuth(HttpBearer):def authenticate(self, request, token):if token == "supersecret":return tokenapi = NinjaAPI(auth=GlobalAuth())

      • 在全局认证时，如果某些函数不需要全局认证，则将该路由路径中的auth设置为None。

      • from ninja import NinjaAPI, Form
from ninja.security import HttpBearerclass GlobalAuth(HttpBearer):def authenticate(self ,request, token):if token == "supersecret":return tokenapi = 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 Clientclass 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 APIKeyHeaderclass 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 APIKeyCookieclass 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 HttpBearerclass 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 HttpBasicAuthclass 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, APIKeyHeaderclass AuthCheck:def authenticate(self, request, key):if key == "supersecret":return keyclass QueryKey(AuthCheck, APIKeyQuery):passclass 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 HttpBearerapi = 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 tokenraise 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 NinjaAPIclass 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 NinjaAPIapi = NinjaAPI(version="1.0.0")@api.get("/hello")
def hello(request):return {"message": "Hello form v1"}

    • # api_v2.py
from ninja import NinjaAPIapi = 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_v2urlpatterns = [...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 Parserclass MyYamlParser(Parser):def parse_body(self, request):return yaml.safe_load(request.body)api = NinjaAPI(parser=MyYamlParser())class Payload(Schema):ints: List[int]string: strf: float@api.post("/yaml")
def operation(request, payload: Payload):return payload.dict()

  • ORJSON 解析器

    • import orjson
from ninja import NinjaAPI
from ninja.parser import Parserclass 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 BaseRenderclass MyRenderer(BaseRender):media_type = "text/plain"def render(self, request, data, *, response_status):return ... # your serialization hereapi = 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 BaseRenderclass 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 BaseRendererclass 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:passelse: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 NinjaAPIapi = NinjaAPI(csrf=True)

  • 注意：将API与基于cookie的身份验证一起使用是不安全！

  • 如果这样做，会报错。

    • from ninja import NinjaAPI
from ninja.security import django_authapi = NinjaAPI(auth=django_auth)

    • 必须启用csrf检查，基于cokkie的身份验证才安全。

      • from ninja import NinjaAPI
from ninja.security import django_authapi = 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 AsyncElasticsearchapi = 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简单教程相关推荐

  1. pypy mysql 兼容_PyPy运行Django+MySQL简单教程

    PyPy是Python开发者为了更好的Hack Python创建的项目.此外,PyPy比CPython是更加灵活,易于使用和试验,以制定具体的功能在不同 PyPy是Python开发者为了更好的Hack ...

  2. python写邮箱系统_Python django实现简单的邮件系统发送邮件功能

    本文实例讲述了Python django实现简单的邮件系统发送邮件功能.分享给大家供大家参考,具体如下: django邮件系统 Django发送邮件官方中文文档 总结如下: 1.首先这份文档看三两遍是 ...

  3. 超实用的Django入门基础教程，新手必备！

    Django入门基础教程 Django简介 安装Django 虚拟环境的建立和准备工作 一.通过pycharm创建 手动创建app 迁移数据库 二.手动创建 创建一个新的文件夹 创建虚拟环境 激活虚拟 ...

  4. Django REST Framework教程(4): 玩转序列化器(Serializer)

    在前面的文章中我们以博客为例,自定义了一个简单的 ArticleSerializer 类, 并分别以函数视图(FBV)和基于类的视图(CBV)编写了博客文章列表资源和单篇文章资源的API,支持客户端以 ...

  5. django 实现简单的搜索功能

    django 实现简单的搜索功能 1 年,1 月前 4849 字 13259 阅读 7 评论 搜索是一个复杂的功能,但对于一些简单的搜索任务,我们可以使用 django model 层提供的一些内置方 ...

  6. Django REST Framework教程(10): 限流(throttle)详解与示例

    在前面的DRF系列教程中,我们以博客为例介绍了序列化器(Serializer), 并使用APIView和ModelViewSet开发了针对文章资源进行增删查改的完整API端点,并详细对权限.认证(含j ...

  7. Django 2入门教程：利用GoormIDE和Bootstrap 4开发旅游博客

    原版书名:<Django 2.1 Tutorial : Build a Travel Blog with GoormIDE and Bootstrap 4 (Tutorial Project)& ...

  8. ASP.NET Core 异常和错误处理 - ASP.NET Core 基础教程 - 简单教程，简单编程

    原文:ASP.NET Core 异常和错误处理 - ASP.NET Core 基础教程 - 简单教程,简单编程 ASP.NET Core 异常和错误处理 上一章节中,我们学习了 ASP.NET Cor ...

  9. idea 从svn导入多个项目_IDEA导入项目简单教程

    该教程用于IDEA初学者导入eclipse项目,或者导入其他已经写好的Java源程序的简单教程. 我们用IDEA打开一个已经写好的项目源文件时,如果没有配置好,就会出现:JDK配置失误报错.程序无法启 ...

最新文章

  1. 教你在Python中用Scikit生成测试数据集（附代码、学习资料）
  2. 项目集成自动分词系统ansj，实现自定义词库
  3. LAMP搭建之三：php编译安装
  4. redis开启redis_Redis聚类
  5. 将可见的电子签名添加到PDF
  6. esc键没反应_有机人名反应——Brown 硼氢化反应（Brown Hydroboration）
  7. springboot 2.x 使用定时任务执行多次问题
  8. 浅谈==和equals
  9. [JavaScript]自定义MessageBox
  10. 解决jquery下checked取值问题...
  11. html5画直线,小白学canvas-从0到1，画直线
  12. asp.net Ajax表单提交 二种方式数据处理 asp.net
  13. php 会员到期提醒_会员管理系统花了50万都没做成，用这五个功能轻松实现
  14. 关键路径c语言,有向图的关键路径的C程序实现代码
  15. 亿能bms上位机_上位机软件上位机监控锂电池管理系统bms软件开发
  16. PDF 文件格式 基本结构
  17. CAD2018下载、安装步骤！
  18. 焦点较中的网络视频相关
  19. Ubuntu系统如何搜索要安装的软件包
  20. Stm32-SWD下载调试配置

热门文章

  1. 云服务器建站原理_第一篇博客---阿里云服务器建站过程（小菜鸟的第一次尝试）...
  2. 阿拉伯数字金额转换为中文大写
  3. 利用VBA代码解决Excel下拉菜单跳过空单元格的问题
  4. flutter实现一个下拉菜单【基于PopupMenuButton】
  5. 四种期权损益结构图，助您轻松理解期权损益
  6. MATLAB基础学习（二）-变量类型与赋值
  7. Chrome游戏《Boxel Rebound》可自制赛道的弹跳小方块
  8. java面试笔试资料 百度网盘
  9. 学习笔记-IP地址2
  10. 基于 Python 的音乐流派分类