Using GraphQL with Python – A Complete Guide
Shadid Haque
We recently released an updated Python tutorial post using Strawberry and GraphOS. You can check it out here!
Known for its ease of use and simplicity, Python is one of the most beloved general-purpose programming languages. And GraphQL, a declarative query language for APIs and server runtimes, pairs quite nicely with Python. Unfortunately, there are very few comprehensive learning materials out there that give you a step-by-step breakdown of how to use GraphQL with Python. This article will go over everything you need to know to get up and running with GraphQL API using Python, Flask, and Ariadne.
You can find the complete code for this post on GitHub.
Learning objectives
By the end of the article, you should know how to:
- Set up a Python web server with Flask
- Use the Ariadne library to implement GraphQL
- Compose a GraphQL Schema
- Perform queries and mutations against a Python GraphQL API
GraphQL vs REST: What problem does GraphQL solve? If you are completely new to GraphQL and want to know how it differs from a traditional REST API, I recommend reading “What is GraphQL? GraphQL introduction“.
Setting up GraphQL with Python (Flask)
Let’s dive into creating our very own GraphQL API with Python. For this demo, we will be using the Flask web server. If you are more accustomed to other frameworks such as Django, you can adapt this codebase to your framework. The basic concepts of GraphQL and Python are more or less the same across various frameworks.
Creating a new python virtual environment
First of all, let’s create a new project and change the directory to the project folder.
mkdir graphql-python-api
cd graphql-python-api
In Python, best practices are to use a virtual environment. We can create a new virtual environment by running the following command.
python3 -m venv myapp
Next, we have to activate the virtual environment. If you are on a Linux or Mac machine you can run the source
command with the path of the activate script like shown below.
source myapp/bin/activate
And if you’re on a windows machine, you can run the following command to activate the virtual environment.
myapp/bin/activate.bat
Installing dependencies
Our application relies on the following dependencies:
- Flask — this is the web server that we’ll use
- Flask-SQLAlchemy — an ORM that makes it easier for us to communicate with our SQL database
- Ariadne — a library for GraphQL python integration
- Flask-Cors — an extension for Cross Origin Resource Sharing
You can install them all using a single command:
pip install flask ariadne flask-sqlalchemy flask-cors
Up and running with a simple Flask app
We will make the following directory structure. The first file we’ll start working with is the api/__init__.py
file, which will hold all the API-related configuration code.
For now, let’s populate the api/__init__.py
with the following code.
from flask import Flask
from flask_cors import CORS
app = Flask(__name__)
CORS(app)
@app.route('/')
def hello():
return 'My First API !!'
Our app.py
file is what’s responsible for actually starting the flask app. Let’s import the flask API instance using the following code:
from api import app
Next, we tell Flask to start the application by looking at our app.py
file. In the command line, we can accomplish this by setting the FLASK_APP
environment variable.
export FLASK_APP=app.py
Finally, we run the app by running the flask run
command.
Great! We can see our Flask app up and running. Before we enable it to use GraphQL, lets hook up a database and define some tables.
Adding a database
For this example, we are going to be using a Postgres DB instance. I like ElephantSQL, a hosted SQL database, but you can use any SQL database you like.
In ElephantSQL, once the instance is provisioned on the cloud, we can see the database server information. If you’re using ElephantSQL, copy the DB URL as shown below, otherwise, copy the URL to where your SQL database is – whether it’s running locally on your machine or with another hosted SQL database service.
We can now add this database url to the __init__.py as shown below.
from flask import Flask
from flask_cors import CORS
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
CORS(app)
app.config["SQLALCHEMY_DATABASE_URI"] = "postgres://mycreds.db.elephantsql.com:5432/ngimluxm"
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
db = SQLAlchemy(app)
@app.route('/')
def hello():
return 'My First API !!'
Restart the server and make sure everything is working as usual.
Creating a model
Next, let’s create our first model.
In our database, we are going to have a Post
table. A Post
will have a unique id
, a title
, description
, and the date
it was created.
Create an api/models.py
file and a new class called Post
as shown below.
from app import db
class Post(db.Model):
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String)
description = db.Column(db.String)
created_at = db.Column(db.Date)
def to_dict(self):
return {
"id": self.id,
"title": self.title,
"description": self.description,
"created_at": str(self.created_at.strftime('%d-%m-%Y'))
}
We can update our app.py
file to include the current models and database settings.
from api import app, db
from api import models
At this point, we can use the Python interactive terminal to create our table and add some records to it. Let’s do that.
First, let’s open the Python terminal by running the following command.
python
Once inside the Python terminal, run the following command to create our table.
>>> from app import db
>>> db.create_all()
On the first line, we import the database instance and on the second, we run the create_all()
method to create related tables based on the model we specified earlier.
Troubleshoot: If you are using Mac for development, you might run into an issue where python can not find psycopg
. To resolve this, run pip install psycopg2-binary
within your virtual environment.
To verify whether the table got created or not, hop into the psql
database console and run the following SQL
query to get the name of all available tables.
SELECT table_name
FROM information_schema.tables
WHERE table_schema='public'
AND table_type='BASE TABLE';
Let’s add a few posts to the Post
table directly from Python command prompt.
>>> from datetime import datetime
>>> from api.models import Post
>>> current_date = datetime.today().date()
>>> new_post = Post(title="A new morning", description="A new morning details", created_at=current_date)
>>> db.session.add(new_post)
>>> db.session.commit()
With a working web API connected to the database, we’re ready to integrate GraphQL into the server.
Writing the GraphQL Schema
A schema in GraphQL describes the shape of our data graph. It is the core of any GraphQL server implementation. It defines the functionality available to the client applications that consumes the API. GraphQL has its own language (GraphQL Schema Definition Language) that is used to write the schema. The schema determines what resources the clients can query and update.
Let’s go ahead and create a new file called schema.graphql
in our root directory. Copy and paste the following code in the file.
schema {
query: Query
}
type Post {
id: ID!
title: String!
description: String!
created_at: String!
}
type PostResult {
success: Boolean!
errors: [String]
post: Post
}
type PostsResult {
success: Boolean!
errors: [String]
post: [Post]
}
type Query {
listPosts: PostsResult!
getPost(id: ID!): PostResult!
}
First of all, we have a schema type defined in the top. This determines what type of operations clients can perform. For now, clients can only perform Query
operations.
Next, observe the Post
type. You will notice that the structure of the Post
type is identical to our Post
model that we defined earlier.
The PostsResult
type defines the structure of the response object when we query for all the posts in the database.
Similarly, PostResult
represents the response when we query for one post in the database.
Finally, we have the type Query.
This type defines the query operations that our clients can perform. Currently, we have two queries: a listPosts
query to grab all the posts from the database, and a getPost
query to get a particular post by its id
.
Wiring up Flask server and GraphQL with Ariadne library
Thus far, we have our Flask server up and running, we connected to a database, and we’ve created our first GraphQL schema. Next, we need to wire up our server with GraphQL, so that we can start using the queries/mutations defined in the schema. We will be using the Ariadne library to do this.
Ariadne is a lightweight Python library that lets you get up and running with GraphQL quickly. Ariadne is framework agnostic (which means you can use it with Flask, Django, or any other framework of your choice) and it uses a schema first approach to GraphQL API development. In this approach, we define our schema first (as we did for this demo app) and write the business logic based on our schema.
Another popular pattern is to use a code first approach while designing GraphQL APIs (Graphene is a popular library that does this). If you’re interested in learning more about this approach, I recommend you give this article a read.
Let’s go and make the following changes in our app.py
file.
from api import app, db
from ariadne import load_schema_from_path, make_executable_schema, \
graphql_sync, snake_case_fallback_resolvers, ObjectType
from ariadne.constants import PLAYGROUND_HTML
from flask import request, jsonify
type_defs = load_schema_from_path("schema.graphql")
schema = make_executable_schema(
type_defs, snake_case_fallback_resolvers
)
@app.route("/graphql", methods=["GET"])
def graphql_playground():
return PLAYGROUND_HTML, 200
@app.route("/graphql", methods=["POST"])
def graphql_server():
data = request.get_json()
success, result = graphql_sync(
schema,
data,
context_value=request,
debug=app.debug
)
status_code = 200 if success else 400
return jsonify(result), status_code
On lines 2 ~ 4, we import a couple functions from the Ariadne library. On line 7, we import the types from our GraphQL schema. Then, we call the make_executable_schema
method from Ariadne. We pass the type definitions as the first argument. The second argument snake_case_fallback_resolvers
is a Bindable; these are special types from Ariadne library that is used to bind python methods to GraphQL schema.
Next, we have two methods. The first method will load up the GraphQL user interface for us. The second method is a POST
method. This endpoint is will be used by our clients to run queries and mutations.
Testing our server
We can run the application by running flask run
.
flask run
Next, we need a GraphQL IDE to build our queries, explore the schema, and test the API functionality. The Apollo Explorer is a free to use GraphQL IDE built specifically for GraphQL developers working on GraphQL APIs. It comes with a lot of powerful features like one-click query building, intelligent search, and a multitude of other productivity features.
To get started, head over to studio.apollographql.com/dev and create an account (using either GitHub or your email). Choose a name for our graph, and select the development
option as the graph type.
We will add our localhost endpoint http://localhost:5000/graphql
in the endpoint field and click create graph.
Once the setup is done, we will see that the GraphQL IDE will load up in our browser.
Query all posts
We are still not able to run queries. Let’s change that. We will write our first query resolver that will return all the posts in the database.
Writing our first Resolver
We can create a new file called api/queries.py
and write the following resolver method as shown below.
from .models import Post
def listPosts_resolver(obj, info):
try:
posts = [post.to_dict() for post in Post.query.all()]
print(posts)
payload = {
"success": True,
"posts": posts
}
except Exception as error:
payload = {
"success": False,
"errors": [str(error)]
}
return payload
This resolver method is very self explanatory. We are trying to query all the Posts from the database and return them as a Payload dictionary.
We have to reference this resolver in our app.py
file. Let’s make the following changes to app.py
file.
from api import app, db
from ariadne import load_schema_from_path, make_executable_schema, \
graphql_sync, snake_case_fallback_resolvers, ObjectType
from ariadne.constants import PLAYGROUND_HTML
from flask import request, jsonify
from api.queries import listPosts_resolver
query = ObjectType("Query")
query.set_field("listPosts", listPosts_resolver)
type_defs = load_schema_from_path("schema.graphql")
schema = make_executable_schema(
type_defs, query, snake_case_fallback_resolvers
)
@app.route("/graphql", methods=["GET"])
def graphql_playground():
return PLAYGROUND_HTML, 200
@app.route("/graphql", methods=["POST"])
def graphql_server():
data = request.get_json()
success, result = graphql_sync(
schema,
data,
context_value=request,
debug=app.debug
)
status_code = 200 if success else 400
return jsonify(result), status_code
On line 6, we are importing the resolver. We are then creating a query instance and specifying the query field and the corresponding resolver (line 9). Finally, we are adding the query instance to the make_executable_schema method call as a parameter. Restart the server, go back to the GraphQL playground and you will be able to run the following query.
query AllPosts {
listPosts {
success
errors
posts {
id
title
description
created_at
}
}
}
Querying a single post by id
Next, we will take a look at how we can query a single item by a property. For this example, we will query a Post
by its id.
We will create a new resolver method inside our queries.py
file.
from ariadne import convert_kwargs_to_snake_case
...
@convert_kwargs_to_snake_case
def getPost_resolver(obj, info, id):
try:
post = Post.query.get(id)
payload = {
"success": True,
"post": post.to_dict()
}
except AttributeError: # todo not found
payload = {
"success": False,
"errors": ["Post item matching {id} not found"]
}
return payload
We imported a decorator called convert_kwargs_to_snake_case
from Ariadne. This decorator converts the method arguments from camel case to snake case.
Let’s update the app.py
file to include the latest resolver
...
from api.queries import listPosts_resolver, getPost_resolver
query = ObjectType("Query")
query.set_field("listPosts", listPosts_resolver)
query.set_field("getPost", getPost_resolver)
...
We can run the query and verify if everything is working as expected.
query GetPost {
getPost(id: "1") {
post {
id
title
description
}
success
errors
}
}
Mutation
Mutations are used to create, update or delete records from the database. Let’s set up our first mutation.
Creating a new post
First of all, in our schema, we need to define the type of mutation we are trying to add. In our case, we want to create a new post. Therefore, we will make a mutation called createPost.
// schema.graphql
schema {
query: Query
mutation: Mutation
}
type Mutation {
createPost(title: String!, description: String!, created_at: String): PostResult!
}
...
We updated our schema.graphql file accordingly as shown above. We add a new Mutation type. We specify the mutation name, required parameters and finally update schema type to include Mutation type.
Updating the schema itself will not do much. We need a resolver to correspond to the createPost
mutation in the schema.
We will create a new file called api/mutations.py
. All our mutation resolvers will live in this file.
# mutations.py
from datetime import date
from ariadne import convert_kwargs_to_snake_case
from api import db
from api.models import Post
@convert_kwargs_to_snake_case
def create_post_resolver(obj, info, title, description):
try:
today = date.today()
post = Post(
title=title, description=description, created_at=today.strftime("%b-%d-%Y")
)
db.session.add(post)
db.session.commit()
payload = {
"success": True,
"post": post.to_dict()
}
except ValueError: # date format errors
payload = {
"success": False,
"errors": [f"Incorrect date format provided. Date should be in "
f"the format dd-mm-yyyy"]
}
return payload
The resolver method is pretty self-explanatory. We are here trying to create and save a new instance of a Post
. On success, we return the post.
We also need to bind this new mutation resolver in our app.py.
...
from api.queries import listPosts_resolver, getPost_resolver
from api.mutations import create_post_resolver
query = ObjectType("Query")
mutation = ObjectType("Mutation")
query.set_field("listPosts", listPosts_resolver)
query.set_field("getPost", getPost_resolver)
mutation.set_field("createPost", create_post_resolver)
type_defs = load_schema_from_path("schema.graphql")
schema = make_executable_schema(
type_defs, query, mutation, snake_case_fallback_resolvers
)
..
As you can see from the code example above, importing and binding the mutation follows the same pattern as importing and binding queries that we have done previously. we can now hop into the GraphQL playground and try to execute this new mutation.
mutation CreateNewPost {
createPost(
title: "New Blog Post",
description:"Some Description") {
post {
id
title
description
created_at
}
success
errors
}
}
Updating a post
Next, we will be looking at updating a post. To do so we will follow the same pattern. First, we will update the schema and add a new mutation called updatePost
.
type Mutation {
createPost(title: String!, description: String!, created_at: String): PostResult!
updatePost(id: ID!, title: String, description: String): PostResult!
}
updatePost
takes in a mandatory parameter id and optional parameters title and description. Now we can create a resolver for this mutation.
# mutations.py
...
@convert_kwargs_to_snake_case
def update_post_resolver(obj, info, id, title, description):
try:
post = Post.query.get(id)
if post:
post.title = title
post.description = description
db.session.add(post)
db.session.commit()
payload = {
"success": True,
"post": post.to_dict()
}
except AttributeError: # todo not found
payload = {
"success": False,
"errors": ["item matching id {id} not found"]
}
return payload
In this method, we are querying the post by id and updating the title and description of the post. We can wire this new resolver up in app.py
like the previous one.
...
from api.mutations import create_post_resolver, update_post_resolver
mutation = ObjectType("Mutation")
mutation.set_field("createPost", create_post_resolver)
mutation.set_field("updatePost", update_post_resolver)
That’s it. We can restart the server and run the updatePost
mutation now.
mutation UpdatePost {
updatePost(id:"2", title:"Hello title", description:"updated description") {
post {
id
title
description
}
success
errors
}
}
Deleting a post
Finally, let’s take a look how we can delete a post. We are going to exactly the same thing as we did with updatePost mutation. We will first create the deletePost mutation in the schema.
type Mutation {
createPost(title: String!, description: String!, created_at: String): PostResult!
updatePost(id: ID!, title: String, description: String): PostResult!
deletePost(id: ID): PostResult!
}
Once that is done we can create a new resolver for it and reference it in the app.py file.
# mutations.py
...
@convert_kwargs_to_snake_case
def delete_post_resolver(obj, info, id):
try:
post = Post.query.get(id)
db.session.delete(post)
db.session.commit()
payload = {"success": True, "post": post.to_dict()}
except AttributeError:
payload = {
"success": False,
"errors": ["Not found"]
}
return payload
# app.py
...
from api.mutations import create_post_resolver, update_post_resolver, delete_post_resolver
...
mutation.set_field("deletePost", delete_post_resolver)
Let’s test the functionality.
Awesome, we now have our GraphQL and Python API up and running.
Final thoughts
The main intention of this article was to get you up and running with GraphQL and Python, as well as introduce some widely used patterns and best practices. I hope you found this article informative.
This is just the start. I suggest checking out some of the other posts on the Apollo blog on topics like caching, GraphQL security, and if you’re really into Python, checking out the rest of the Ariadne documentation.
That’s a wrap! Happy hacking and see you next time.