Instantly share code, notes, and snippets.

What would you like to do?
import graphene
from graphene import String
from graphene_sqlalchemy import SQLAlchemyObjectType, SQLAlchemyConnectionField
from promise import Promise
from promise.dataloader import DataLoader
from sqlalchemy import Column, BigInteger, ForeignKey
from sqlalchemy.orm import relationship
from db import Base, db_session # 这里自己去定义
# 定义Model
class UserModel(Base):
__tablename__ = "users"
id = Column(BigInteger, primary_key=True)
name = Column(String(255))
posts = relationship("PostModel", backref="posts")
class PostModel(Base):
__tablename__ = "posts"
id = Column(BigInteger, primary_key=True)
user_id = Column(BigInteger, ForeignKey(""))
# 定义schema
class Post(SQLAlchemyObjectType):
class Meta:
model = PostModel
class User(SQLAlchemyObjectType):
posts = graphene.List(Post)
def resolve_posts(self, info, **args):
return (
.load( # 即是DataLoader里面的key
.then(lambda response: [response])
class Meta:
model = UserModel
# 定义Connection Field
class UsersConnectionField(SQLAlchemyConnectionField):
def __init__(self, type, *args, **kwargs):
super().__init__(type, uuid=String(), *args, **kwargs)
# 重写默认的get_query方法以支持自定义参数
def get_query(cls, model, info, sort=None, **args):
query = super().get_query(model, info, None, **args)
if "limit" in args:
query = query.limit(args["limit"])
if "offset" in args:
query = query.offset(args["offset"])
return query
# 定义DataLoader
class PostsDataLoader(DataLoader):
def batch_load_fn(self, keys):
q = db_session.query(PostModel).filter(PostModel.uuid.in_(keys))
posts = dict([(, post) for post in q.all()])
return Promise.resolve([posts.get(id, None) for id in keys])
class Query(graphene.ObjectType):
users = graphene.List(User, limit=graphene.Int(), offset=graphene.Int())
def resolve_users(self, info, **args):
query = UsersConnectionField.get_query(UserModel, info, None, **args)
return query.all()
schema = graphene.Schema(query=Query)
if __name__ == "__main__":
query = """
query {
users (limit:10, offset:20) {
posts {
result = schema.execute(
context_value={"session": db_session, "post_data_loader": PostsDataLoader()},
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment