Flask框架-Graphql的使用 – 潘登同学的flask学习笔记

Graphql

官网：https://graphql.cn/

GraphQL与RESTful关系

• GraphQL和RESTful一样，都是一种网站架构，一种前后端通信规范，不涉及语言，不同语言有不同的实现方案。
• GraphQL目前被认为是革命性的API工具，因为它可以让客户端在请求中指定希望得到的数据，而不像传统的RESTful那样只能呆板地在服务端进行预定义。
• 这样它就让前、后端团队的协作变得比以往更加的通畅，从而能够让组织更好地运作。
• 而实际上，GraphQL与RESTful都是基于HTTP进行数据的请求与接收，而且GraphQL也内置了很多RESTful模型的元素在里面。
• 那么在技术层面上，GraphQL和RESTful这两种API模型到底有什么异同呢？他们归根到底其实没多大区别，只不过GraphQL做了一些小改进，使得开发体验产生了较大的改变。

相同点

• 都有资源这个概念，而且都能通过ID去获取资源
• 都可以通过HTTP GET方式来获取资源
• 都可以使用JSON作为响应格式

差异点

• 在RESTful中，你所访问的路径就是该资源的唯一标识（ID）；在GraphQL中，该标识与访问方式并不相关
• 在RESTful中，资源的返回结构与返回数量是由服务端决定；在GraphQL，服务端只负责定义哪些资源是可用的，由客户端自己决定需要得到什么资源
• 如果是对资源的简单查询，GraphQL与RESTful是类似的，都是通过指定资源的名称以及相关参数来取得，但不同的是，你可以根据资源之间的关联关系来发起一个复杂请求，而在RESTful中你只能定义一些特殊的URL参数来获取到特殊的响应，或者是通过发起多个请求、再自行把响应得到的数据进行组装才行。RESTful对数据的描述形式是一连串的URL端点，而GraphQL则是由相互之间有所关联的schema组成。

路由处理器（Route Handlers）vs 解析器（Resolvers）

这是一个Restful的路由代码

class HelloWorld(Resource):def get(self):return {'say': 'hello world'}api.add_resource(HelloWorld, '/hello')

这里我们得到了一个可以返回“world”这个字符串的/hello端点。从这个例子我们可以看到一个RESTful API请求的的生命周期：

• 服务器收到请求并提取出HTTP方法名（比如这里就是GET方法）与URL路径
• API框架找到提前注册好的、请求路径与请求方法都匹配的代码
• 该段代码被执行，并得到相应结果
• API框架对结果进行序列化，添加上适当的状态码与响应头后，返回给客户端

而这是一个GraphQL的代码

class Query(ObjectType):"""定义一个字符串属性域hello""" say = String()def resolve_say(root,info):return  f'Hello World!'

我们看到，这里并没有针对某个URL路径提供函数，而是把Query类型中的hello字段映射到一个函数上了。

在GraphQL中这样的函数我们称之为解析器（Resolver）。

然后我们就可以这样发起一个查询：

query {say
}

至此，总结一下服务器对一个 GraphQL 请求的执行过程：

• 服务器收到 HTTP 请求，取出其中的 GraphQL 查询
• 遍历查询语句，调用里面每个字段所对应的 Resolver。在这个例子里，只有 Query 这个类型中的一个字段 hello
• Resolver 函数被执行并返回相应结果
• GraphQL 框架把结果根据查询语句的要求进行组装因此我们将会得到如下响应

{ "say": "Hello world!" }

还可以多次调用同一个Resolvers

query {say
secondSay: say
}

形象表达为:

GraphQL的使用（Python版）

官方文档 https://graphql.cn/code/#python

安装GraphQL

pip install graphene

第一个GraphQL程序

from graphene import ObjectType,Schema,String# 查询用的
class Query(ObjectType):say = String()def resolve_say(self,info):return 'Hello GraphQL!!'if __name__ == '__main__':schema = Schema(query=Query)# 定义查询query_string = '''{say}'''# 执行查询rs = schema.execute(query_string)print(rs.data)

FLaks中应用GraphQL–graphene

pip install Flask-GraphQL

from flask import Flask
from flask_graphql import GraphQLView
from graphene import ObjectType,Schema,String# 创建对象
app = Flask(__name__)
# 路由地址
@app.route("/")def index():return "pandeng"class Query(ObjectType):hello = String()def resolve_hello(self,info):return 'Hello FLask GraphQL!!'schema = Schema(query=Query)
# 与挂载类试图相似
app.add_url_rule('/graphql',view_func=GraphQLView.as_view('graphql',schema=schema,graphiql=True))if __name__ == '__main__':app.run(debug=True)

报错解决

cannot import name 'get_default_backend' from 'graphql

pip install graphql-core==2.2.1
pip install graphene==2.1.8

ERROR: pip's dependency resolver does not currently take into account all the packages that are installed. This behaviour is the source of the following dependency conflicts.graphql-server-core 1.2.0 requires graphql-core<3,>=2.3, but you have graphql-core 2.2.1 which is incompatible.graphql-relay 3.1.0 requires graphql-core>=3.1, but you have graphql-core 2.2.1 which is incompatible.graphene 3.0 requires graphql-core~=3.1.2, but you have graphql-core 2.2.1 which is incompatible.

这一段的报错就是全部重装，然后最后再执行一下上面的代码,不要用pip来重装，到site-package中，把与GraphQL有关的全删了；

graphene属性参数

from flask import Flask
from flask_graphql import GraphQLView
from graphene import ObjectType,Schema,String# 创建对象
app = Flask(__name__)
# 路由地址
@app.route("/")def index():return "pandeng"class Query(ObjectType):hello = String()# 需求 将输出的数据的JSON对象的键改为my_info(也会改调用的方法名)，性别默认为男，个人描述为必填项me = String(name='my_info',sex=String(default_value='男'),desc = String(required=True))def resolve_hello(self,info):return 'Hello FLask GraphQL!!'# self是一个None,而info是一个解析器，记住这两个是必填项就行def resolve_me(self,info,sex,desc):# 函数这里设置的参数要与上面命名的对应上return f'PD，sex:{sex}, desc:{desc}'schema = Schema(query=Query)
app.add_url_rule('/graphql',view_func=GraphQLView.as_view('graphql',schema=schema,graphiql=True))if __name__ == '__main__':app.run(debug=True)

最终结果

数据类型

• graphene.String:字符串
• graphene.Int:整数
• graphene.Float:浮点
• graphene.Boolean:布尔值
• graphene.ID:唯一的字符串
• graphene.Data:年月日
• graphene.Time:时分秒
• graphene.DateTime:年月日时分秒
• graphene.Decimal:钱
• graphene.JSONString:JSON字符串
• graphene.NonNull:数据返回不可能为空
• graphene.List:列表

基本数据类型

import graphene
from flask import Flask
from flask_graphql import GraphQLView
import datetime
import decimal
class Query(graphene.ObjectType):str_= graphene.String()int_ = graphene.Int()float_ = graphene.Float()boolean_ = graphene.Boolean()id_ = graphene.ID()def resolve_str_(self,info):return 'Hello'def resolve_int_(self,info):return 1def resolve_float_(self,info):return 1.1def resolve_boolean_(self,info):return Truedef resolve_id_(self,info):return 1.2data_ = graphene.Date()time_ = graphene.Time()date_time = graphene.DateTime()decimal_ = graphene.Decimal()def resolve_data_(self,info):return datetime.date(2050,1,2)def resolve_time_(self,info):return datetime.time(1,2,3)def resolve_date_time(self,info):return datetime.datetime(2050,1,2,3,4,5)def resolve_decimal_(self,info):return decimal.Decimal("10.30")json_ = graphene.JSONString()def resolve_json_(self,info):return {"name":"graphql"}
if __name__ =='__main__':schema = graphene.Schema(query= Query)app = Flask(__name__)app.add_url_rule('/graphql',view_func=GraphQLView.as_view('grapql',schema=schema, graphiql=True))app.run(debug=True)

NonNull与List

class Query(graphene.ObjectType):uname = graphene.String(required=True)# NonNull中不能是方法，也就是不能使graphene.String()user_name = graphene.NonNull(graphene.String) # Exception: NonNull could not have a mounted String() as inner type. Try with NonNull(String).uname_list = graphene.List(graphene.String)uname_list2 = graphene.List(graphene.NonNull(graphene.String))uname_list3 = graphene.NonNull(graphene.List(graphene.String))def resolve_uname(self,info):return Nonedef resolve_user_name(self,info):return Nonedef resolve_uname_list(self,info):return Nonedef resolve_uname_list2(self,info):return [None]def resolve_uname_list3(self,info):return [None]

自定义数据类型

import graphene
from flask import Flask
from flask_graphql import GraphQLView# 记得继承graphene.ObjectType
class Person(graphene.ObjectType):first_name = graphene.String()last_name = graphene.String()class Query(graphene.ObjectType):person = graphene.Field(Person)persons = graphene.List(Person)def resolve_person(self,info):# return Person(first_name="p",last_name="d")# 上下两种方法都可以return {"first_name":"p","last_name":"d"}def resolve_persons(self,info):# return [Person(first_name="p",last_name="d")#  Person(first_name="d",last_name="p")]# 上下两种方法都可以return [{"first_name":"尚学堂","last_name":"Mr.Gao"},{"first_name":"尚学堂","last_name":"Mr.Li"}]
if __name__ =='__main__':schema = graphene.Schema(query= Query)app = Flask(__name__)app.add_url_rule('/graphql',view_func=GraphQLView.as_view('grapql',schema=schema, graphiql=True))app.run(debug=True)

GraphQL接口

这个接口是，工厂方法模式（Factory method pattern），是一种实现了"工厂"概念的面向对象设计模式。就像其他创建型模式一样，它也是处理在不指定对象具体类型的情况下创建对象的问题。工厂方法模式的实质是"定义一个创建对象的接口，但让实现这个接口的类来决定实例化哪个类。工厂方法让类的实例化推迟到子类中进行。

之前的python基础中也实现过工厂模式,这里还是看JAVA怎么说吧 https://zhuanlan.zhihu.com/p/110419316

我们来实现一个创建动物的工厂模式，复习以下元类与类装饰器

'''动态创建类 元类-->类用途动态创建类使用：类 = type(类名,(父类...),{属性,方法})'''
Person = type('Person',(),{})
p1 = Person()
print(p1)##静态创建类
class Animal():def __init__(self,color):self.color = colordef eat(self):print('我要吃手手')def sleep(self):print('我要睡觉觉')Dog = type('Dog',(Animal,),{'age':3,'sleep':sleep})
dahuang = Dog('yellow')
print(dahuang.age)
dahuang.sleep()#%%类装饰器
'''在不修改函数源代码的前提下，增加新的功能
'''
class AAA():def __init__(self,func):# print('我是AAA.init()')self.__func = funcdef __call__(self,*args,**kwargs):self.addfunc()self.__func()def addfunc(self):print('新增用户权限验证')print('新增系统日志处理')
@AAA
#类装饰器相当于   test = AAA(test)
def test():print('我是功能一')test()

然后我们用graphene的方式来实现工厂模式，设计动物类，然后产生老鼠和鸟的类

import graphene
from flask import Flask
from flask_graphql import GraphQLViewclass Animal(graphene.Interface):id = graphene.ID()name = graphene.String()class Mouse(graphene.ObjectType):# Class Meta与type的操作基本差不多，功能差不多class Meta:interfaces =(Animal,)run = graphene.String()class Bird(graphene.ObjectType):class Meta:interfaces =(Animal,)fly = graphene.String()class Query(graphene.ObjectType):mouse = graphene.Field(Mouse)bird = graphene.Field(Bird)animal = graphene.Field(Animal,type_ = graphene.Int(required = True))def resolve_mouse(self,info):# 在自定义数据类型中测试了可以这样returnreturn {'id':1,'name':'杰瑞','run':'跑呀~Tom来了！！'}def resolve_bird(self,info):return {'id':2,'name':'鹦鹉','fly':'跑呀~tina来了！！'}# 返回元类的时候不能笑之前那样写，要先创建一个子类，然后返回去时，就只会返回元类的方法属性等def resolve_animal(self,info,type_):# return Animal(id=3,name='动物') #  "message": "An Interface cannot be intitialized",if type_ == 1:return Mouse(id= 4, name='米老鼠',run='追唐老鸭啦~~~')else:return Bird(id = 3,name='鹦鹉',fly = '跑呀~Tom来了！！')if __name__ =='__main__':schema = graphene.Schema(query= Query)app = Flask(__name__)app.add_url_rule('/graphql',view_func=GraphQLView.as_view('grapql',schema=schema, graphiql=True))app.run(debug=True)

枚举的使用

沿用上面的代码

class Role(graphene.Enum):JR = 1YW = 2TM = 3# 修改一下属性参数
animal = graphene.Field(Animal,type_ =Role())
# 修改一下resolve_animal()
def resolve_animal(self,info,type_):if type_ == Role.JR:return Mouse(id= 4, name='米老鼠',run='追唐老鸭啦~~~')elif type_ == Role.YW:return Bird(id = 3,name='鹦鹉',fly = '跑呀~Tom来了！！')else:return Mouse(id= 5,name="Tom",run="别让杰瑞跑了~~快追！！")

GraphQL查询

先创建数据，数据大概是这样的: 元类是一个角色，子类是人类和机器人;然后有一个枚举类型

'''
===================== data =========================
上面是创建数据的操作
'''
human_data = {}
droid_data = {}def setup():global human_data, droid_dataluke = Human(id="1000",name="Luke Skywalker",friends=["1002", "1003", "2000", "2001"],appears_in=[4, 5, 6],home_planet="Tatooine",)vader = Human(id="1001",name="Darth Vader",friends=["1004"],appears_in=[4, 5, 6],home_planet="Tatooine",)han = Human(id="1002",name="Han Solo",friends=["1000", "1003", "2001"],appears_in=[4, 5, 6],home_planet=None,)leia = Human(id="1003",name="Leia Organa",friends=["1000", "1002", "2000", "2001"],appears_in=[4, 5, 6],home_planet="Alderaan",)tarkin = Human(id="1004",name="Wilhuff Tarkin",friends=["1001"],appears_in=[4],home_planet=None,)human_data = {"1000": luke,"1001": vader,"1002": han,"1003": leia,"1004": tarkin,}c3po = Droid(id="2000",name="C-3PO",friends=["1000", "1002", "1003", "2001"],appears_in=[4, 5, 6],primary_function="Protocol",)r2d2 = Droid(id="2001",name="R2-D2",friends=["1000", "1002", "1003"],appears_in=[4, 5, 6],primary_function="Astromech",)droid_data = {"2000": c3po, "2001": r2d2}def get_character(id):return human_data.get(id) or droid_data.get(id)def get_friends(character):return map(get_character, character.friends)def get_hero(episode):if episode == 5:return human_data["1000"]return droid_data["2001"]def get_human(id):return human_data.get(id)def get_droid(id):return droid_data.get(id)'''
===================== schema =========================
下面是类的设置与查询设置
'''import graphene
from flask_graphql import GraphQLView
from flask import Flaskclass Episode(graphene.Enum):  # 剧集NEWHOPE = 4  # 星球大战4 新希望EMPIRE = 5  # 黑金帝国JEDI = 6    # 星球大战 绝地class Character(graphene.Interface):  # 角色 元类id = graphene.ID()name = graphene.String()friends = graphene.List(lambda: Character)appears_in = graphene.List(Episode)     # 出演def resolve_friends(self, info):# 返回朋友的一个列表# The character friends is a list of stringsreturn [get_character(f) for f in self.friends]class Human(graphene.ObjectType):   # 人类class Meta:interfaces = (Character,)home_planet = graphene.String()  # 地球家园class Droid(graphene.ObjectType):   # 机器人class Meta:interfaces = (Character,)primary_function = graphene.String()    # 主要功能class Query(graphene.ObjectType):# 查询参数只有三个hero = graphene.Field(Character, episode=Episode())human = graphene.Field(Human, id=graphene.String())droid = graphene.Field(Droid, id=graphene.String())def resolve_hero(root, info, episode=None):return get_hero(episode)def resolve_human(root, info, id):return get_human(id)def resolve_droid(root, info, id):return get_droid(id)if __name__ == '__main__':setup()schema = graphene.Schema(query=Query)app = Flask(__name__)app.add_url_rule('/graphql', view_func=GraphQLView.as_view('graphql',schema=schema, graphiql=True))app.run(debug=True)

• 1.查一个，查询语句

human(id:"1002"){idnamefriends {idname}appearsIn}

• 2.当我们想在一个界面查询多个数据时，会报错

{human(id:"1002"){idnamefriends {id}appearsIn}human(id:"1003"){idnamefriends {id}appearsIn}
}

按照他的提示，给他起个别名,结果正常

{h1:human(id:"1002"){idnamefriends {id}appearsIn}h2:human(id:"1003"){idnamefriends {id}appearsIn}
}

• 3.查询中简化查询语句，结果一致

# 将查询语句封装成一个类 起个名字hf
fragment hf on Human{idnamehomePlanet
}
# 进行查询
{h1:human(id:"1002"){...hf}h2:human(id:"1003"){...hf}
}

• 4.之前我们查元类的时候，只会返回元类特有的属性，如果我们想查询子类也有的属性值,加...on

{hero(episode:EMPIRE){idnamefriends {id}appearsIn# 想查询homePlanet，但是hero是元类，没有homePlanet属性# 但是hero(episode:EMPIRE)返回的是一个id为1000的人类，人类是有homePlanet属性的# 增加一个...on Human{}...on Human{homePlanet}...on Droid{primaryFunction# 能正常查但是没结果}}
}

• 5.给查询语句定义名称,动态传递参数，在查询下面输入参数

query q1($id:Episode) {# $id理解为变量，Episode理解为参数hero(episode:$id){idnamefriends {id}appearsIn...on Droid{primaryFunction}...on Human{homePlanet}}
}

从这一套的操作的时候，就能发现，GraphQL是需要什么才查什么，就是比Restful更加灵活

GraphQL增删改

使用Mutation关键字,增加数据的代码是比较难理解，注释写的很清晰了，一些很难理解的就是固定写法了，核心其实就是创建数据，存储数据，返回相应结果；复杂的略过即可。

import graphene
from flask import Flask
from flask_graphql import GraphQLViewusers =[]
# 第0步：数据的定义
class Person(graphene.ObjectType):name = graphene.String()age = graphene.Int()# 第二步创建增加数据的函数
class CreatePerson(graphene.Mutation):# 定义可以增加的属性，类似查询参数时的 hero = graphene.Field(Character, episode=Episode())person = graphene.Field(Person)msg = graphene.String()class Arguments:    # 设置可以传进来的参数name = graphene.String()age = graphene.Int()# 接受传进来的参数，类似查询时候传进来的查询参数，而这里是增加数据而已def mutate(self,info,name,age):# 存储的逻辑p = Person(name =name,age = age)# 这里没有连数据库，直接用的本地listusers.append(p)# 响应请求的结果 返回查询结果return CreatePerson(person = p,msg="success")# Mutation 第一步
class Mutation(graphene.ObjectType):    # 映射对外使用字段# 增加数据 的函数create_person = CreatePerson.Field()class Query(graphene.ObjectType):persons = graphene.List(Person)def resolve_persons(self,info):return users
if __name__ =='__main__':schema = graphene.Schema(query=Query,mutation=Mutation)app = Flask(__name__)app.add_url_rule('/graphql',view_func=GraphQLView.as_view('grapql',schema=schema, graphiql=True))app.run(debug=True)

增加语句

mutation add1{createPerson(age:19,name:"pandeng"){person {nameage}msg}
}

注意 在GraphQL中其实是前端在查询，所以字符串要双引号，不然报错

可以看出虽然写的很复杂，但是最后调用的时候还是很舒服的

import graphene
from flask import Flask
from flask_graphql import GraphQLViewclass Person(graphene.ObjectType):name = graphene.String()age = graphene.Int()users =[Person(name="sxt",age="18"),Person(name="Mr.Li",age= "20")
]class UpdatePerson(graphene.Mutation):person = graphene.Field(Person)msg = graphene.String()class Arguments:name = graphene.String()age = graphene.Int()def mutate(self,info,name,age):for u in users:if u.name == name:u.age = agereturn UpdatePerson(person =u,msg ="success")return UpdatePerson(msg='fail')class DeletePerson(graphene.Mutation):person = graphene.Field(Person)msg = graphene.String()class Arguments:name = graphene.String()def mutate(self,info,name):for u in users:if u.name == name:users.remove(u)return UpdatePerson(person =u,msg ="success")return UpdatePerson(msg='fail')# Mutation
class Mutation(graphene.ObjectType):    # 映射对外使用字段# 这里可以理解为注册到Mutation上，就像之前注册到Query上一样update_person = UpdatePerson.Field()delete_person = DeletePerson.Field()class Query(graphene.ObjectType):persons = graphene.List(Person)def resolve_persons(self,info):return users
if __name__ =='__main__':schema = graphene.Schema(query=Query,mutation=Mutation)app = Flask(__name__)app.add_url_rule('/graphql',view_func=GraphQLView.as_view('grapql',schema=schema, graphiql=True))app.run(debug=True)

GraphQL连接数据库SQLAlchemy

pip install graphene-sqlalchemy

新建一个数据库

CREATE DATABASE GraphQL_dbDEFAULT CHARACTER SET = 'utf8mb4';

先创建数据

from random import randint
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Column, ForeignKey,Integer,String,and_,Text,Table,func,create_engine# 数据库的变量
HOST = '127.0.0.1'  # 自己本机的就是127.0.0.1 或 localhost
PORT = 3306
DATA_BASE = 'GraphQL_db'
USER = 'root'
PWD = 'xxx'
#  # DB_URI = f'数据库的名+驱动名://{USER}:{PWD}@{HOST}:{PORT}/{DATA_BASE}'
DB_URI = f'mysql+pymysql://{USER}:{PWD}@{HOST}:{PORT}/{DATA_BASE}'
engine = create_engine(DB_URI)
# 创建一个基础类
Base = declarative_base(engine)Session = sessionmaker(engine)class User(Base):__tablename__ = "t_user"id = Column(Integer,primary_key=True,autoincrement=True)uname = Column(String(50),nullable=False)age = Column(Integer)# newss = relationship('News')  不推荐def __repr__(self) -> str:return f'<User: id={self.id} uname={self.uname} age={self.age}>'def create_data():Base.metadata.create_all()with Session() as session:for i in range(10):user = User(uname=f'uname{i}',age=randint(5,20))session.add(user)session.commit()if __name__ == '__main__':create_data()

Base.query = scoped_session(Session).query_property()关键 ，先贴上源码:

接上面代码

from random import randint
import graphene
from graphene_sqlalchemy import SQLAlchemyObjectType
from flask import Flask
from flask_graphql import GraphQLView# 这句话才能是允许查询
Graph_Session = scoped_session(Session)
Base.query = Graph_Session.query_property()class graphene_user(SQLAlchemyObjectType):# 继承了SQLAlchemyObjectType 就可以直接继承之前创建的User了class Meta:model = Userclass Query(graphene.ObjectType):# 注册 查询所有users = graphene.List(graphene_user)user= graphene.Field(graphene_user,id=graphene.String())# 解析def resolve_users(self,info):query = graphene_user.get_query(info)return query.all()def resolve_user(self,info,id):return graphene_user.get_node(info,id)if __name__ =='__main__':schema = graphene.Schema(query=Query)# 这样可以测试接口query_with_argument = 'query qi{user(id:"3"){uname,id,age}}'result = schema.execute(query_with_argument)print(result.data['user'])app = Flask(__name__)app.add_url_rule('/graphql',view_func=GraphQLView.as_view('grapql',schema=schema, graphiql=True))app.run(debug=True)

查询结果:

新增数据

from random import randint
import graphene
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker,scoped_session
from sqlalchemy import Column, ForeignKey,Integer,String,and_,Text,Table,func,create_engine
from graphene_sqlalchemy import SQLAlchemyObjectType
from flask import Flask
from flask_graphql import GraphQLView# 数据库的变量
HOST = '127.0.0.1'  # 自己本机的就是127.0.0.1 或 localhost
PORT = 3306
DATA_BASE = 'GraphQL_db'
USER = 'root'
PWD = 'xxx'
#  # DB_URI = f'数据库的名+驱动名://{USER}:{PWD}@{HOST}:{PORT}/{DATA_BASE}'
DB_URI = f'mysql+pymysql://{USER}:{PWD}@{HOST}:{PORT}/{DATA_BASE}'
engine = create_engine(DB_URI)
# 创建一个基础类
Base = declarative_base(engine)
Session = sessionmaker(engine)
# 这句话才能是允许查询
Graph_Session = scoped_session(Session)
Base.query = Graph_Session.query_property()class User(Base):__tablename__ = "t_user"id = Column(Integer,primary_key=True,autoincrement=True)uname = Column(String(50),nullable=False)age = Column(Integer)# newss = relationship('News')  不推荐def __repr__(self) -> str:return f'<User: id={self.id} uname={self.uname} age={self.age}>'def create_data():Base.metadata.drop_all()Base.metadata.create_all()with Session() as session:for i in range(10):user = User(uname=f'uname{i}',age=randint(5,20))session.add(user)session.commit()class graphene_user(SQLAlchemyObjectType):# 继承了SQLAlchemyObjectType 就可以直接继承之前创建的User了class Meta:model = Userclass CreateUser(graphene.Mutation):user = graphene.Field(graphene_user)msg = graphene.Boolean()# 指定参数class Arguments:# id = graphene.String() # 不传了因为是自增长uname = graphene.String()age = graphene.Int()# 解析def mutate(self,info,uname,age):user = User(uname=uname,age=age)# 加入数据库# 使用with Session形式报错，无法响应到前端，但是能添加数据# Instance <User at 0x1eb97368a60> is not bound to a Session; # attribute refresh operation cannot proceed# with Session() as session:#     session.add(user)#     session.commit()Graph_Session.add(user)Graph_Session.commit()return CreateUser(user=user,msg=True)class Mutation(graphene.ObjectType):# 注册添加一个create_user = CreateUser.Field()class Query(graphene.ObjectType):# 注册 查询所有users= graphene.List(graphene_user)# 注意前面导包导进来的是sqlalchemy的String，而这里要用GraphQL的stringuser= graphene.Field(graphene_user,id=graphene.String())# 解析def resolve_users(self,info):query = graphene_user.get_query(info)return query.all()def resolve_user(self,info,id):return graphene_user.get_node(info,id)if __name__ =='__main__':# create_data()schema = graphene.Schema(query=Query,mutation=Mutation)app = Flask(__name__)app.add_url_rule('/graphql',view_func=GraphQLView.as_view('grapql',schema=schema, graphiql=True))app.run(debug=True)

对于修改和删除

就是把前面的修改删除拿下来改吧改吧就ok了

报错解决

Instance <User at 0x1eb97368a60> is not bound to a Session; attribute refresh operation cannot proceed

错误代码,不能正常响应，但是可以添加数据到数据库

class CreateUser(graphene.Mutation):user = graphene.Field(graphene_user)msg = graphene.Boolean()# 指定参数class Arguments:# id = graphene.String() # 不传了因为是自增长uname = graphene.String()age = graphene.Int()# 解析def mutate(self,info,uname,age):user = User(uname=uname,age=age)# 加入数据库# 使用with Session形式报错，无法响应到前端，但是能添加数据with Session() as session:session.add(user)session.commit()return CreateUser(user=user,msg=True)

原因： 过早关闭session，（我也不知道，之前这样做都是可以的），修改为

class CreateUser(graphene.Mutation):user = graphene.Field(graphene_user)msg = graphene.Boolean()# 指定参数class Arguments:# id = graphene.String() # 不传了因为是自增长uname = graphene.String()age = graphene.Int()# 解析def mutate(self,info,uname,age):user = User(uname=uname,age=age)# 加入数据库# 修改后!!!!!!!!!!!!!!!!!!!!!!!!Graph_Session.add(user)Graph_Session.commit()return CreateUser(user=user,msg=True)

使用graphene_sqlalchemy

from random import randint
import graphene
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker,scoped_session
from sqlalchemy import Column, ForeignKey,Integer,String,and_,Text,Table,func,create_engine
from graphene_sqlalchemy import SQLAlchemyConnectionField, SQLAlchemyObjectType
from flask import Flask
from flask_graphql import GraphQLView# 数据库的变量
HOST = '127.0.0.1'  # 自己本机的就是127.0.0.1 或 localhost
PORT = 3306
DATA_BASE = 'GraphQL_db'
USER = 'root'
PWD = 'mysql2002'
#  # DB_URI = f'数据库的名+驱动名://{USER}:{PWD}@{HOST}:{PORT}/{DATA_BASE}'
DB_URI = f'mysql+pymysql://{USER}:{PWD}@{HOST}:{PORT}/{DATA_BASE}'
engine = create_engine(DB_URI)
# 创建一个基础类
Base = declarative_base(engine)
Session = sessionmaker(engine)
# 这句话才能是允许查询
Graph_Session = scoped_session(Session)
Base.query = Graph_Session.query_property()class User(Base):__tablename__ = "t_user"id = Column(Integer,primary_key=True,autoincrement=True)uname = Column(String(50),nullable=False)age = Column(Integer)# newss = relationship('News')  不推荐def __repr__(self) -> str:return f'<User: id={self.id} uname={self.uname} age={self.age}>'def create_data():Base.metadata.drop_all()Base.metadata.create_all()with Session() as session:for i in range(10):user = User(uname=f'uname{i}',age=randint(5,20))session.add(user)session.commit()class graphene_user(SQLAlchemyObjectType):# 继承了SQLAlchemyObjectType 就可以直接继承之前创建的User了class Meta:model = User# 接口 继承graphene.relay.Node# 因为Query是一个通用的查询，可以查询很多类，所以只要继承了这个接口，就能用他的方法查询Interfaces = (graphene.relay.Node,)class Query(graphene.ObjectType):node = graphene.relay.Node.Field()user = SQLAlchemyConnectionField(graphene_user.connection)if __name__ =='__main__':# create_data()schema = graphene.Schema(query=Query)app = Flask(__name__)app.add_url_rule('/graphql',view_func=GraphQLView.as_view('grapql',schema=schema, graphiql=True))app.run(debug=True)

规范操作后出现null

然后我们的数据库中也是有数据的

那怎么办？ 去看源码，点进Node中，因为Node是接口元类

那我们就把刚才的源码中提到的type name:id(这里的name:id,nama是指查询的那个子类，id是主键)去加一波密，在网站中输入base64，进行编码，结果为Z3JhcGhlbmVfdXNlcjoxMQ==

重新查询，查询成功

node查询总结

虽然这样查询比较繁琐，但是全都集成进了node节点，真正实现了一个节点进行查询，实现了这幅图的效果;

前端就使用base64的库将参数进行传递加密再进行传递即可；

后续再增加数据，就只需要再写一个model(比如User)对应的子类graphene_model(比如graphene_user),一样继承Node节点，最后挂载在Query上即可；

graphene_sqlalchemy中connection查询

• first:前几个
• last：后几个 last:x
• after：索引x 往后(不含x) after：arrayconnection: x
• before:索引x 往前(不含x)

刚才我们挂载的时候写的一条语句似乎还没有用上，我们去使用一下

class Query(graphene.ObjectType):node = graphene.relay.Node.Field()user = SQLAlchemyConnectionField(graphene_user.connection)

注意到右边还有一些参数，好像能排序，试试年龄升序

还能返回前几个数据和后几个数据 ,first与last

看到还有befor与after，在一通翻源码下，我们知道了还是一个base64的编码arrayconnection:

把arrayconnection:4拿去转码，转为base64(YXJyYXljb25uZWN0aW9uOjQ=，放进before后面与after后面

组合操作

需求：对应python切片[2:5]

组合操作，after:1 first:3 或者 before:5 first:3

me:id去加一波密，在网站中输入base64，进行编码，结果为Z3JhcGhlbmVfdXNlcjoxMQ==`

组合操作，after:1 first:3 或者 before:5 first:3

Flask框架-Graphql的使用相关推荐

1. python的flask框架显示柱状图_使用Python的Flask框架，结合Highchart，动态渲染图表(Ajax 请求数据接口)...

   参考链接:https://www.highcharts.com.cn/docs/ajax 参考链接中的示例代码是使用php写的,这里改用python写. 需要注意的地方: 1.接口返回的数据格式,这个 ...

2. Python 使用 Flask框架记录

   Python 使用 Flask框架记录 1.安装Flask ​ Flask依赖两个外部库,Werkzeug和Jinja2,Werkzeug是一个WSGI(服务器网关接口).Jinja2时负责渲染模板. ...

3. flask框架如何实现修改密码和免密登录功能

   flask是python web开发的常用框架之一.本文将讲述flask如何实现修改密码和免密登录功能 修改密码功能 数据库部分: #重置密码 def reset_pass(phone,passwor ...

4. 浅谈Python flask框架浅析

   前言 Python 面向对象的高级编程语言,以其语法简单.免费开源.免编译扩展性高,同时也可以嵌入到C/C++程序和丰富的第三方库,Python运用到大数据分析.人工智能.web后端等应用场景上. P ...

5. Flask框架-基本使用

   Flask简介 Web应用程序的本质 Web(World Wide Web)诞生最初的目的,是为了利用互联网交流工作文档. 为什么要用Web框架? 如何生产一辆汽车? web网站发展至今,特别是服务器 ...

6. python断点调试不起作用flask_关于python的flask框架下的调试

   一开始我用的vs code,单纯的python程序可以调试,没有问题,但是一旦是flask框架的py程序,vs code出现bug了,调试不了,具体表现在端口会变成默认的5000端口,断点不停,变量不 ...

7. python之flask框架详解

   目录 前言 1.新建文件helloworld.py 2.相关配置参数 3.加载配置文件 3.1配置对象加载 3.2配置文件加载 3.3读取配置 4 路由定义 4.1指定路由地址 4.2 给路由传参 4 ...

8. html文件怎么用Python做后端,利用python实现后端写网页（flask框架）.pdf

   利利用用python实实现现后后端端写写网网页页 ((flask框框架架)) 如何用python做后端写网页-flask框架 什么是Flask安装flask模块Hello World更深一步:数据绑 ...

9. python 框架好学吗-python的flask框架难学吗

   Flask框架难学吗?它和Django哪个更容易一些,这可能是学Python web开发的同学经常问的问题,下面来说一下flask框架. Flask是python的web框架,最大的特征是轻便,让开发 ...

最新文章

1. OpenCV3.3中 K-最近邻法(KNN)接口简介及使用
2. 第 5 章 Nova - 030 - Launch和Shut Off操作详解
3. SpringBoot配置属性之DataSource
4. c语言编写程序数一下 1到100的所有整数中出现多少次数字9
5. VMProtect SDK完全避坑指南
6. 从源码角度看Android系统SystemServer进程启动过程
7. JavaScript实现数除以二divideByTwo算法（附完整源码）
8. 服务器数据库怎么添加计划维护,服务器数据库怎么添加计划维护
9. nodejs接收post数据
10. 递归应用：八皇后问题
11. Java中大数值是什么？数值想存多大存多大
12. 国家电网人工智能专利数量排名竟然靠前，腾讯，百度都不如国家电网
13. python基础教程-Python入门教程完整版（懂中文就能学会）
14. Java Error（四）
15. python的标准库——turtle
16. OCJP认证有什么用
17. 2011年11月份第二周51Aspx源码发布详情
18. 数据库笔试题 (3)
19. 加密狗映射至虚拟服务器,ESXI 5.1/5.5 主机添加或映射USB设备（加密狗）(示例代码)...
20. Joey Logano 嘉年华正式开始，和 Joey Logano 一起竞赛吧

热门文章

1. python制作app windows_windows桌面app自动化，python也可以
2. 科威特政府成功测试诺基亚XGS-PON技术 速度高达10Gbps
3. Vue 使用 vis-network 绘制网络关系图
4. Java 读取资源文件 项目根路径
5. 插入u盘需重启计算机,量产后的U盘一插入电脑就提示重启的原因及解决方法
6. linux服务器，卸载tensorflow CPU 安装PGU版
7. 一文读懂 DeFi 自动做市商崛起原因、尚存局限和未来前景
8. 假如，张东升是个程序员（神结尾）
9. C语言的三套标准：C89、C99和C11
10. cortex-m4启动代码解读