API#

ArkID基于Django-ninja框架来开发API，完整继承了其所有能力。

自定义API#

可以通过Django-ninja与 Django原生 两种方式来自定义API。

主要区别在于，Django-ninja方式定义会自动出现在openapi.json中,并依赖Schema，Django原生则不会或不需要。

Django-ninja 的 API 定义方式#

使用 arkid.core.extension.Extension.register_api

创建Schema#

使用 arkid.core.extension.create_extension_schema

注意！该函数的第二个参数，是指的插件的 __init__.py 文件所在的目录

示例

from arkid.core import extension

class CaseExtension(extension.Extension): 
    def load(self): 
        super().load()

        self.register_api('/path/', 'GET', self.api_func)

    def api_func(self, request):
        pass

权限#

arkid对于API权限的控制，是通过定义角色的方式实现的，arkid支持4种角色分别为普通用户(NORMAL_USER)、租户管理员(TENANT_ADMIN)、平台管理员(PLATFORM_ADMIN)、平台用户(PLATFORM_USER)。通过在具体的方法装饰器上引入对应角色，可以实现对应角色的用户对于接口的访问

权限

...
from arkid.core.constants import NORMAL_USER, TENANT_ADMIN, PLATFORM_ADMIN, PLATFORM_USER #引入角色
from arkid.core.api import operation #引入操作装饰器
...


@api.get("/tenant/{tenant_id}/apps/", response=List[AppListItemOut], tags=['应用'])
@operation(AppListOut, roles=[TENANT_ADMIN, PLATFORM_ADMIN]) # 引入了租户管理员身份，平台管理员身份
@paginate(CustomPagination)
def list_apps(request, tenant_id: str):
    '''
    公开app列表
    '''
    pass
    return []

分页#

arkid提供基础分页器功能，其使用方法如下：

分页

...
from ninja.pagination import paginate #引入分页装饰器
from arkid.core.pagenation import CustomPagination #引入分页器
...


# 声明返回列表项结构
class AppGroupListItemOut(Schema):
    id:str
    name:str
# 声明返回结构体
class AppGroupListOut(ResponseSchema):
    data: List[AppGroupListItemOut]

@api.get("/path/", response=List[AppGroupListItemOut]) #注意 此处因分页器会自动封装错误提示等数据  故而此处不需要填写封装错误信息后的Schema
@operation(AppGroupListOut)
@paginate(CustomPagination)
def get_app_groups(request,tenant_id: str):
    """ 应用分组列表
    """
    groups = AppGroup.expand_objects.filter(tenant__id=tenant_id)
    parent_id = query_data.dict().get("parent_id",None)
    groups = groups.filter(parent__id=parent_id)
    return groups.all()

Django 的 API 定义方式#

使用 arkid.core.extension.Extension.register_routers

示例

from arkid.core import extension
from django.urls import re_path
from django

class CaseExtension(extension.Extension): 
    def load(self): 
        super().load()

        class CaseView(View):
            def post(self,request):
                pass

        path_list = [
            re_path(rf'^/path/$',self.api_func),
            re_path(rf'^/path2/$',CaseView.as_view()),
        ]
        self.register_routers(path_list)

    def api_func(self,request):
        pass

修改内核API#

必要的时候，我们需要更改原内核中的API。

修改request#

修改request，就是修改API相关的Request Schema.

之后，我们希望获取该request，并执行自定义的逻辑.

在每个API响应之前，都会抛出一个事件，事件tag为: operation_id + '_pre', 侦听该事件即可获取request对象

提示

operation_id 可以在 openapi.json 中查找

使用

示例

from arkid.core import extension
from api.v1.views.app import AppConfigSchemaIn

class CaseExtension(extension.Extension): 
    def load(self): 
        super().load()

        self.register_extend_api(AppConfigSchemaIn, case1=str, case2=(str, Field(title='case2_name')))
        self.listen_event('api_v1_views_app_list_apps_pre',self.app_list_pre_handler)

    def app_list_pre_handler(self,event,**kwargs):
        print(event.request.case1)
        print(event.request.case2)

修改response#

修改response，除了要修改Response Schema以外，还需要真的改变最终的返回值

在每个API响应完毕之后，也会抛出一个事件，事件的tag为 operation_id，侦听该事件并修改事件中的reponse即可。

使用

示例

from arkid.core import extension
from api.v1.views.app import AppConfigSchemaOut

class CaseExtension(extension.Extension): 
    def load(self): 
        super().load()

        self.register_extend_api(AppConfigSchemaOut, case1=str, case2=(str, Field(title='case2_name')))
        self.listen_event('api_v1_views_app_list_apps',self.app_list_handler)

    def app_list_pre_handler(self,event,**kwargs):
        event.response['case1'] = 'case1'
        event.response['case2'] = 'case2'