graphene-django 开发

GraphQL 是一个用于 API 的查询语言，是一个使用基于类型系统来执行查询的服务端运行时（类型系统由你的数据定义）。GraphQL 并没有和任何特定数据库或者存储引擎绑定，而是依靠你现有的代码和数据支撑。

一个 GraphQL 服务是通过定义类型和类型上的字段来创建的，然后给每个类型上的每个字段提供解析函数`

graphql 和 django 结合开发

pip install graphene_django

官方

如果，只是熟悉django，可以先熟知graphene之后再次提升。

API入口

my_project/my_project/urlls.py

pip install graphene-file-upload

from django.contrib import admin
from django.urls import re_path, path

from graphene_file_upload.django import FileUploadGraphQLView

urlpatterns = [
    path('admin/', admin.site.urls),
  	# 使用FileUploadGraphQLView原因是上传文件，必须使用这个view。否则无法上传。
    path('api/graphql/', FileUploadGraphQLView.as_view(graphiql=True)),
]

my_project/my_project/schema.py


import graphene
import account.schema
import work_hour_management.schema


# 查询注册->主
class Query(
    account.schema.Query,
    work_hour_management.schema.Query,
    graphene.ObjectType,
):
    ...


# 增删改注册->主
class Mutation(
    account.schema.Mutation,
    work_hour_management.schema.Mutation,
    graphene.ObjectType,
):
    ...

# 对应settings.py中 GRAPHENE
schema = graphene.Schema(query=Query, mutation=Mutation)

配置

GRAPHENE = {
    'SCHEMA': 'core.schema.schema',
  	# graphne_jwt 配置
    'MIDDLEWARE': [
        'graphql_jwt.middleware.JSONWebTokenMiddleware'
    ]
}

认证配置

pip install django-graphql-jwt

认证相关配置

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    # 第三方
    'graphene_django',
    'graphql_jwt.refresh_token.apps.RefreshTokenConfig',
]

# 用户模型
AUTH_USER_MODEL = 'account.User'
# jwt 配置
GRAPHQL_JWT = {
    'JWT_VERIFY_EXPIRATION': True,
    'JWT_LONG_RUNNING_REFRESH_TOKEN': True,
    'JWT_EXPIRATION_DELTA': timedelta(minutes=1000 if DEBUG else 10),
    'JWT_REFRESH_EXPIRATION_DELTA': timedelta(days=15),
    'JWT_AUTH_HEADER_PREFIX': "Bearer",  # default 'JWT'
    'JWT_AUTH_HEADER_NAME': 'HTTP_AUTHORIZATION'
}
# 认证
AUTHENTICATION_BACKENDS = [
    "graphql_jwt.backends.JSONWebTokenBackend",
    "django.contrib.auth.backends.ModelBackend",
]

模块开发

schema

/my_project/account/schema.py

graphene schema.py应用类似djangourl。将所有的接口注册到Query或者 Mutation。最后统一注册到my_project/my_project/schema.py

class Query(
    UserQuery,
):
    ...
    
class Mutation(graphene.ObjectType):
    token_auth = ObtainJSONWebToken.Field()
    verify_token = graphql_jwt.Verify.Field()
    refresh_token = graphql_jwt.Refresh.Field()
    create_user = CreateUser.Field()

查询

my_project/模块/query/

如果模块较大可以单独拆出来一个文件夹query或者直接使用query.py

# 使用DjangoObjectType
class SystemQuery(graphene.ObjectType):
	
    select = graphene.List(
        SystemType,
        args={
            'key': graphene.Int(required=True, description='标识')
        },
        description='公共下拉框'
    )

    @staticmethod
    @AuthDecorators.login_required
    def resolve_select(root, info, **kwargs):
        return System.objects.filter(is_delete=False, category=kwargs.get('key'))

在类中写接口，类似drf中的类视图中写CRUD，区别是graphene是一个类中写无数接口。最好不要这么做！每个类作为一个模块，将所有相关接口放进去是一个比较好的选择。

增删改

基础mutation

class BaseClientIDMutation(graphene.relay.ClientIDMutation):
    error_message = graphene.String(default_value="")
    ok = graphene.Boolean(default_value=True)

    class Meta:
        abstract = True

    @classmethod
    def mutate_and_get_payload(cls, root, info, **kwargs):
      	# 类中捕捉异常或者在perform_mutation返回需要通知给客户端的信息
        try:
            result = cls.perform_mutation(root, info, **kwargs)
        except Exception as e:
            if settings.DEBUG:
                return cls(error_message=traceback.format_exc(), ok=False)
            else:
                return cls(error_message=e, ok=False)
        else:
            return result

    @classmethod
    def perform_mutation(cls, root, info, **kwargs):
        """
        Here is you mutation logic
        """
        raise NotImplementedError

class CreateUser(BaseClientIDMutation):
    class Input:
        department_id = graphene.ID(description='部门', required=True)
        position_id = graphene.ID(description='岗位', required=True)
        role_list = graphene.List(graphene.String, description='角色', required=True)
        leader_id = graphene.ID(description='leader')
        induction_at = graphene.String(description='入职时间', required=True)
        mobile = graphene.String(description='手机号', required=True)
        name = graphene.String(description='姓名', required=True)
        username = graphene.String(description='用户名', required=True)

    @classmethod
    @AuthDecorators.login_required
    @check_permission(permission_code='user_create')
    def perform_mutation(cls, root, info, **kwargs):
        var, data = verify_data(**kwargs)
        if not var:
            return cls(**data)
        try:
            User.objects.get(username=kwargs['username'])
            return cls(error_message='用户名已存在', ok=False)
        except User.DoesNotExist:
            data['username'] = kwargs['username']
        with atomic():
            role_list = data.pop('role_list')
            user = User.objects.create(**data)
            user.set_password(data['mobile'])
            user.save()
            # 设置员工角色
            user.role.add(*role_list)

        return cls(error_message='', ok=True)

types

有点类似drf中的序列化器。直接使用DjangoObjectType

class UserType(DjangoObjectType):
    class Meta:
        model = User

基础ObjectType

class UserType(graphene.ObjectType):
    name = graphene.String()

分页

pagination

使用django的pagintaion封装

import graphene
from django.db.models import QuerySet
from django.core.paginator import Paginator
from typing import Union


class PaginationBase:
    length = graphene.Int()


class PaginationField(graphene.Field):

    def __init__(self, of_type, *args, **kwargs):
        # 设置分页默认字段
        kwargs.setdefault(
            "page",
            graphene.Int(
                required=False,
                description="当前页码，必须和per_page一起使用"
            )
        )
        kwargs.setdefault(
            "per_page", graphene.Int(
                required=False,
                description="每页数量，必须和page一起使用"
            )
        )
        super().__init__(of_type, *args, **kwargs)

    @staticmethod
    def paging(
            queryset: QuerySet or list,
            per_page: Union[int, None] = None,
            current_page: Union[int, None] = None
    ) -> "dict":
        result = {
            "data": [],
            "length": 0,
        }

        if per_page and current_page:
            paginator = Paginator(queryset, per_page)
            result["data"] = paginator.get_page(current_page)
            result["length"] = paginator.count
        else:
            # 不做分页处理，返回全部
            result["data"] = queryset
            result["length"] = queryset.count()

        return result

    @classmethod
    def paging_queryset(cls, func):
        def wrapper(*args, **kwargs):
            queryset = func(*args, **kwargs)
            return cls.paging(
                queryset,
                current_page=kwargs.get("page"),
                per_page=kwargs.get("per_page")
            )

        return wrapper

types

class UserTypes(DjangoObjectType):
    class Meta:
        model = User
        interfaces = (graphene.relay.Node,)
        fields = ('id', 'name')
        
class PaginationUser(PaginationBase, graphene.ObjectType):
    data = graphene.List(UserTypes)

query

user_list = PaginationField(
    PaginationUser,
    args={
        'department': graphene.ID(required=False, description='部门'),
    },
    description='员工分页'
)

@staticmethod
@AuthDecorators.login_required
@PaginationField.paging_queryset
def resolve_user_list(root, info, **kwargs):
    return User.objects.all()

返回格式

{
  "data": [{}, {}],
  "length": 2
}

秒客网

graphene-django 开发

目录

API入口

配置

认证配置

模块开发

schema

查询

增删改

types

分页

pagination

types

query

相关文章