Alex Chamberlain, Team Leader @ Bloomberg LP
@alexchamberlain
Slides: https://alexchamberlain.github.io/presentation-pylondinium-graphql-in-python/
Example Code: https://github.com/alexchamberlain/city-api/
Inspire you to investigate writing GraphQL servers in Python, including sign posting you towards open source libraries available.
GraphQL is a query language for your API, and a server-side runtime for executing queries by using a type system you define for your data. GraphQL isn’t tied to any specific database or storage engine and is instead backed by your existing code and data.
query country {
country(iso: "US") {
iso
name
cities {
geonameID
name
}
}
}
query city_by_id {
london: city(geonameID: 2643743) {
geonameID
name
latitude
longitude
population
country {
iso
name
}
}
new_york: city(geonameID: 5128581) {
geonameID
name
latitude
longitude
country {
iso
name
}
}
}
def default_resolve_fn(source, resolve_info, **kwargs):
name = resolve_info.field_name
property = getattr(source, name, None)
if callable(property):
return property()
return property
23 repositories on graphql-python organisation on GitHub
type Country {
iso: String!
name: String!
cities: [City!]!
}
type City {
geonameID: Int!
name: String!
latitude: Float!
longitude: Float!
population: Int!
country: Country!
}
type Query {
country(iso: String): Country!
city(name: String, geonameID: Int): City!
}
schema {
query: Query
}
from aiodataloader import DataLoader
from .models import Country
class CountryLoader(DataLoader):
def __init__(self, db):
super().__init__()
self.db = db
async def batch_load_fn(self, isos):
query = 'SELECT iso, name FROM country WHERE iso = any($1::text[])'
rows = await self.db.fetch(query, isos)
mapping = {
row['iso']: Country(
iso=row['iso'],
name=row['name']
) for row in rows
}
return [mapping.get(i) for i in isos]
def resolver(self, context, resolve_info, *, iso):
return self.load(iso)
def context_resolver(self, context, resolve_info):
return self.load(context.country_iso)
def build_schema():
data = pkgutil.get_data('cities', 'cities.graphql').decode()
schema = make_executable_schema(
parse_schema(data),
{
'Query': {
'country': loader_resolver('Country', CountryLoader.resolver),
'city': city_resolver
},
'Country': {
'cities': country_cities_resolver
},
'City': {
'country': loader_resolver('Country', CountryLoader.context_resolver)
}
}
)
return schema
from graphql.execution.executors.asyncio import AsyncioExecutor
from sanic import Sanic
import sanic_graphql
@app.listener('before_server_start')
async def init_graphql(app, loop):
schema = build_schema()
app.add_route(
GraphQLView.as_view(
schema=schema,
executor=AsyncioExecutor(loop=loop),
graphiql=True
),
'/graphql',
methods=['POST', 'OPTIONS']
)
class GraphQLView(sanic_graphql.GraphQLView):
def get_context(self, request):
return {
'db': request['db'],
'loaders': {
'Country': CountryLoader(request['db']),
'City': CityLoader(request['db'])
}
}