Skip to content

Instantly share code, notes, and snippets.

@Bachmann1234
Created April 25, 2020 20:37
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save Bachmann1234/8fd34bc1dac6d33b78fccf82ee263177 to your computer and use it in GitHub Desktop.
Save Bachmann1234/8fd34bc1dac6d33b78fccf82ee263177 to your computer and use it in GitHub Desktop.
Example graphql server
import uvicorn
from graphene import (
ObjectType,
String,
Schema,
AbstractType,
Field,
Int,
List,
Mutation,
Boolean,
)
from starlette.applications import Starlette
from starlette.routing import Route
from starlette.graphql import GraphQLApp
DATABASE = {
"users": [{"id": 1, "name": "Bachmann"}, {"id": 2, "name": "Rachel"}],
"connections": [
{"id": 1, "user_id": 1, "friend_id": 2},
{"id": 2, "user_id": 2, "friend_id": 1},
],
"posts": [{"id": 1, "author_id": 1, "content": "Hello World!"}],
"comments": [{"id": 1, "post_id": 1, "author_id": 2, "comment": "Bit cliche..."}],
}
def _find_by_field(field, field_value, table):
for row in table:
if row[field] == field_value:
return row
raise ValueError("Not Found")
def _find_by_id(value, table):
return _find_by_field("id", value, table)
class User(ObjectType):
id = String()
name = String()
connections = List(lambda: Connection)
def resolve_connections(parent, info):
return [
connection
for connection in DATABASE["connections"]
if connection["user_id"] == parent["id"]
]
class Connection(ObjectType):
id = String()
user = Field(User)
friend = Field(User)
def resolve_user(parent, info):
return _find_by_id(parent["user_id"], DATABASE["users"])
def resolve_friend(parent, info):
return _find_by_id(parent["friend_id"], DATABASE["users"])
class Post(ObjectType):
id = String()
author = Field(User)
content = String()
comments = List(lambda: Comment)
def resolve_author(parent, info):
return _find_by_id(parent["author_id"], DATABASE["users"])
def resolve_comments(parent, info):
return [
comment
for comment in DATABASE["comments"]
if comment["post_id"] == parent["id"]
]
class Comment(ObjectType):
id = String()
post = Field(Post)
author = Field(User)
comment = String()
def resolve_post(parent, info):
return _find_by_field(parent["post_id"], DATABASE["posts"])
def resolve_author(parent, info):
return _find_by_id(parent["author_id"], DATABASE["users"])
class CommentQuery(AbstractType):
comments = List(Comment, post_id=Int())
def resolve_comments(parent, info, post_id):
return [
comment for comment in DATABASE["comments"] if comment["post_id"] == post_id
]
class PostQuery(AbstractType):
post = Field(Post, author_id=Int())
def resolve_post(parent, info, author_id):
return _find_by_field("author_id", author_id, DATABASE["posts"])
class ConnectionQuery(AbstractType):
connection = Field(Connection, user_id=Int())
def resolve_connection(parent, info, user_id):
return _find_by_field("user_id", user_id, DATABASE["connections"])
class UserQuery(AbstractType):
user = Field(User, user_id=Int())
def resolve_user(parent, info, user_id):
return _find_by_id(user_id, DATABASE["users"])
class CreatePost(Mutation):
class Arguments:
user_id = String()
post_content = String()
ok = Boolean()
post = Field(lambda: Post)
def mutate(root, info, user_id, post_content):
post = {
"id": max([post["id"] for post in DATABASE["posts"]]) + 1,
"author_id": user_id,
"content": post_content,
}
DATABASE["posts"].append(post)
ok = True
return CreatePost(post=post, ok=ok)
class PostMutations(AbstractType):
create_post = CreatePost.Field()
class Mutations(PostMutations, ObjectType):
pass
class Query(UserQuery, ConnectionQuery, PostQuery, CommentQuery, ObjectType):
pass
schema = Schema(query=Query, mutation=Mutations)
routes = [Route("/graphql", GraphQLApp(schema=schema))]
app = Starlette(routes=routes)
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
@Bachmann1234
Copy link
Author

venv ❯ pip freeze
aniso8601==7.0.0
click==7.1.1
graphene==2.1.8
graphql-core==2.3.1
graphql-relay==2.0.1
h11==0.9.0
httptools==0.1.1
promise==2.3
Rx==1.6.1
six==1.14.0
starlette==0.13.3
uvicorn==0.11.3
uvloop==0.14.0
websockets==8.1

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment