Created
April 25, 2020 20:37
-
-
Save Bachmann1234/8fd34bc1dac6d33b78fccf82ee263177 to your computer and use it in GitHub Desktop.
Example graphql server
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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