Created
November 22, 2015 14:24
-
-
Save ptz0n/465eca6fa2ab1161fbbd to your computer and use it in GitHub Desktop.
This file contains hidden or 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 graphene | |
from graphene import relay | |
USER_DATA = [ | |
{ | |
'user_id': '1', | |
'first_name': 'Peter', | |
'last_name': 'Cabbage', | |
'email_address': '[email protected]', | |
'age': 32 | |
}, | |
{ | |
'user_id': '2', | |
'first_name': 'Lukas', | |
'last_name': 'Chard', | |
'email_address': '[email protected]', | |
'age': 54 | |
}, | |
{ | |
'user_id': '3', | |
'first_name': 'Marie', | |
'last_name': 'Cauliflower', | |
'email_address': '[email protected]', | |
'age': 27 | |
}, | |
] | |
def fetch_user_from_database(user_id): | |
user, = [user for user in USER_DATA if user['user_id'] == user_id] | |
return user | |
def fetch_users_from_database(args): | |
full_name = args.get('fullName', None) | |
if full_name: | |
full_name = full_name.upper() | |
sort_key = args.get('sortKey', None) | |
sort_direction = args.get('sortDirection', None) | |
reversed = False | |
if sort_directon and sort_directon == 'DESC': | |
reversed = True | |
filtered_users = [user for user in USER_DATA if full_name is None or full_name in | |
user['first_name'].upper() + ' ' + user['last_name'].upper()] | |
if sort_key and sort_direction: | |
filtered_users.sort(lambda item: item[sort_key], reverse=reversed) | |
return filtered_users | |
schema = graphene.Schema() | |
class User(relay.Node): | |
"""A User fetched from the database""" | |
user_id = graphene.ID() | |
first_name = graphene.String() | |
last_name = graphene.String() | |
full_name = graphene.String(description='A field created by setting the first and last name.') | |
email_address = graphene.String() | |
age = graphene.Int() | |
@classmethod | |
def get_node(cls, user_id, info): | |
user_dict = fetch_user_from_database(user_id) | |
# user_dict will contain the fields, user_id, first_name, last_name, email_address and age | |
user_dict['id'] = user_dict['user_id'] # will be used to set the global ID used by relay | |
return User(**user_dict) | |
def resolve_full_name(self, *args): | |
return ' '.join([self.first_name, self.last_name]) | |
def _user_container(user): | |
# user_dict will contain the fields, user_id, first_name, last_name, email_address and age | |
user['id'] = user['user_id'] | |
return User(**user) | |
# This will be our root query | |
class Query(graphene.ObjectType): | |
users = graphene.List(User, | |
fullName=graphene.String(), | |
sortKey=graphene.String(), | |
sortDirection=graphene.String()) | |
user = relay.NodeField(User) | |
viewer = graphene.Field('self') # needed for Relay | |
# args will be a dict with 'fullName', 'sortKey' and 'sortDirection' | |
# info is an object with information about the query being sent | |
def resolve_users(self, args, info): | |
list_of_user_dicts = fetch_users_from_database(args) | |
return [_user_container(user) for user in list_of_user_dicts] | |
def resolve_viewer(self, *args, **kwargs): | |
return self | |
# Here we set the root query for our schema | |
schema.query = Query | |
# --- | |
result = schema.execute('query { users { fullName } }') | |
# result.data should be {'users': [{fullName: 'Peter Cabbage'}, {fullName: 'Lukas Chart'}, {fullName: 'Marie Cauliflower'}]} | |
users = result.data['users'] | |
print(users) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment