django使用graphql的实例

时间：2022-02-22 05:39:17

一、开发环境

1、python3.6

2、django2.0

3、window10

二、项目搭建

1、创建一个虚拟空间mkvirtualenv 空间名

2、创建一个django项目

3、安装graphql的依赖包

pip install graphene-django

4、创建一个组件blog

5、把组件blog及graphene_django注入到app中

6、在settings.py中配置mysql数据库连接

三、书写blog的内容

1、在models.py中写上数据模型

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from django.db import models
 
# Create your models here.
class User(models.Model):
 name = models.CharField(max_length=100, verbose_name="博主名字")
 gender = models.CharField(max_length=6, choices=(('male', u'男'), ('female', '女')), default='female',
        verbose_name='性别')
 create_at = models.DateTimeField(auto_now_add=True, verbose_name='创建时间')
 
class Blog(models.Model):
 title = models.CharField(max_length=100, verbose_name='标题')
 user = models.ForeignKey(User, null=True, blank=True, on_delete=models.SET_NULL, verbose_name='博主名字')
 content = models.TextField(verbose_name='博客内容')
 create_at = models.DateTimeField(auto_now_add=True, verbose_name='创建时间')
 update_at = models.DateTimeField(auto_now=True, verbose_name='更新时间')

2、新建一个schema.py文件

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
#!/usr/bin/env python
# encoding: utf-8
 
import graphene
from graphene_django.types import DjangoObjectType
from .models import User, Blog
 
class UserType(DjangoObjectType):
 class Meta:
  model = User
 
class BlogType(DjangoObjectType):
 class Meta:
  model = Blog
 
# 定义动作约素输入类型
class UserInput(graphene.InputObjectType):
 name = graphene.String(required=True)
 gender = graphene.String(required=True)
 
class BlogInput(graphene.InputObjectType):
 title = graphene.String(required=True)
 user = graphene.Int(required=True)
 content = graphene.String(required=True)
 
# 定义一个创建user的mutation
class CreateUser(graphene.Mutation):
 # api的输入参数
 class Arguments:
  user_data = UserInput(required=True)
 
 # api的响应参数
 ok = graphene.Boolean()
 user = graphene.Field(UserType)
 
 # api的相应操作，这里是create
 def mutate(self, info, user_data):
  user = User.objects.create(name=user_data['name'], gender=user_data['gender'])
  ok = True
  return CreateUser(user=user, ok=ok)
 
 
# 定义一个创建博客的mutation
class CreateBlog(graphene.Mutation):
 class Arguments:
  blog_data = BlogInput(required=True)
 
 blog = graphene.Field(BlogType)
 
 def mutate(self, info, blog_data):
  # 插入到数据库中
  blog = Blog.objects.create(title=blog_data['title'], user_id=blog_data['user'], content=blog_data['content'])
  return CreateBlog(blog=blog)
 
# 定义一个查询语句
class Query(object):
 all_user = graphene.List(UserType)
 all_blog = graphene.List(BlogType)
 
 def resolve_all_user(self, info, **kwargs):
  # 查询所有book的逻辑
  return User.objects.all()
 
 def resolve_all_blog(self, info, **kwargs):
  # 查询所有title的逻辑
  return Blog.objects.all()

3、在跟目录(和settings.py同级)创建一个项目的总schema.py

?
1
2
3
4
5
6
7
8
9
10
11
12
13
import graphene
import book.schema, blog.schema
 
class Query(blog.schema.Query, graphene.ObjectType):
 # 总的Schema的query入口
 pass
 
class Mutations(graphene.ObjectType):
 # 总的Schema的mutations入口
 create_user = blog.schema.CreateUser.Field()
 create_blog = blog.schema.CreateBlog.Field()
 
schema = graphene.Schema(query=Query, mutation=Mutations)

4、配置url地址

?
1
2
3
4
5
6
7
8
from django.contrib import admin
from django.urls import path
from graphene_django.views import GraphQLView
from .schema import schema
urlpatterns = [
 path('admin/', admin.site.urls),
 path('graphql/', GraphQLView.as_view(graphiql=True, schema=schema)),
]

5、生成数据库映射及启动项目,直接在浏览器*问

四、可以对上面的代码调整

1、把Mutations也单独定义在各自的schema.py中

?
1
2
3
4
# 定义一个总的mutation出口
class Mutation(graphene.AbstractType):
 create_user = CreateUser.Field()
 create_blog = CreateBlog.Field()

2、在总的schema.py中引入类型Query一样的操作

?
1
2
3
class Mutations(blog.schema.Mutation, graphene.ObjectType):
 # 总的Schema的mutations入口
 pass

3、输入数据类型可以直接定义在mutation里面

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
class CreateUser(graphene.Mutation):
 # api的输入参数(类名可以随便定义)
 class Arguments:
  name = graphene.String(required=True)
  gender = graphene.String(required=True)
 
 # api的响应参数
 ok = graphene.Boolean()
 user = graphene.Field(UserType)
 
 # api的相应操作，这里是create
 def mutate(self, info, name, gender):
  user = User.objects.create(name=name, gender=gender)
  ok = True
  return CreateUser(user=user, ok=ok)

五、Query语句中使用条件查询

1、app的schema(官方案例)

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
import graphene
from graphene_django.types import DjangoObjectType
from .models import Category, Ingredient
 
class CategoryType(DjangoObjectType):
 class Meta:
  model = Category
 
class IngredientType(DjangoObjectType):
 class Meta:
  model = Ingredient
 
# 定义一个查询
class Query(object):
 # 定义一个根据id或者name查询的
 category = graphene.Field(CategoryType,
        id=graphene.Int(),
        name=graphene.String())
 # 查询全部的
 all_categories = graphene.List(CategoryType)
 # 根据条件查询
 ingredient = graphene.Field(IngredientType,
        id=graphene.Int(),
        name=graphene.String())
 # 查询全部的
 all_ingredients = graphene.List(IngredientType)
 
 def resolve_all_categories(self, info, **kwargs):
  return Category.objects.all()
 
 def resolve_all_ingredients(self, info, **kwargs):
  # We can easily optimize query count in the resolve method
  return Ingredient.objects.select_related('category').all()
 
 # 定义查询语句
 def resolve_category(self, info, **kwargs):
  id = kwargs.get('id')
  name = kwargs.get('name')
 
  if id is not None:
   return Category.objects.get(pk=id)
 
  if name is not None:
   return Category.objects.get(name=name)
 
  return None
 
 def resolve_ingredient(self, info, **kwargs):
  id = kwargs.get('id')
  name = kwargs.get('name')
 
  if id is not None:
   return Ingredient.objects.get(pk=id)
 
  if name is not None:
   return Ingredient.objects.get(name=name)
 
  return None

官网地址

补充知识：记录下python中使用定时器的几种方法

方式一、直接使用while循环的方式

?
1
2
3
4
5
6
7
8
9
10
from datetime import datetime
import time
 
# 每n秒执行一次
def timer(n):
  while True:
    print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    time.sleep(n)
 
timer(5)

方式二、使用threading模块中的Timer

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from datetime import datetime
from threading import Timer
 
# 打印时间函数
def print_time(inc):
  print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
  """
  Timer的参数说明
  inc:表示时间间隔
  print_time:执行的函数
  (inc,):传递给执行函数的参数
  """
  t = Timer(inc, print_time, (inc,))
  t.start()
 
print_time(2)

方式三、使用sched模块

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import time
import sched
from datetime import datetime
 
# 初始化 sched 模块的 scheduler 类
# 第一个参数是一个可以返回时间戳的函数，第二个参数可以在定时未到达之前阻塞。
schedule = sched.scheduler(time.time, time.sleep)
 
# 被周期性调度触发的函数
def print_time(inc):
  print(datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
  schedule.enter(inc, 0, print_time, (inc,))
 
# 默认参数 60 s
def start(inc=60):
  # enter四个参数分别为：间隔事件、优先级（用于同时间到达的两个事件同时执行时定序）、被调用触发的函数、给触发函数的参数（tuple形式）
  schedule.enter(0, 0, print_time, (inc,))
  schedule.run()
 
if __name__ == "__main__":
  start(10)

方式四、使用apscheduler

?
1
2
3
4
5
6
7
8
9
10
11
12
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
 
 
def job():
  print(datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
 
 
if __name__ == "__main__":
  scheduler = BlockingScheduler()
  scheduler.add_job(job, 'interval', seconds=5)
  scheduler.start()

以上这篇django使用graphql的实例就是小编分享给大家的全部内容了，希望能给大家一个参考，也希望大家多多支持服务器之家。

原文链接：https://blog.csdn.net/kuangshp128/article/details/79370351

标签：

相关文章