Zero to App Store in 1 day — Create a GraphQL API with Flask, SQLAlchemy & PostgreSQL
I had an app idea a few weeks ago and thought the build would be pretty simple. Having built a few apps before, I already had some code that I would be copying over. After doing some planning, I realized I could build and ship the app in a few days. It took me about a week but I shipped it. The process was so painless and delightful that I thought I’d share the tech stack.
This post will be about getting the backend running. I’ll also publish a second post about building a react native app that connects to this backend. It will be a simple jokes app where users can sign up, create jokes, and view all the jokes created by all the users.
We will use:
- Flask — A python web framework to handle the networking and web requests.
- PostgreSQL — Database to store our data. This could easily be switched to your choosing (mongo, sqlite, mysql).
- SQLAlchemy — An ORM to connect, query, and update our database. I love SQLAlchemy. It’s very powerful once you start writing complicated queries but not as easy as Rails or Django.
- Graphene —A python GraphQL library. Gone are the days of REST, we’re doing GraphQL now.
- Alembic — A lightweight database migration tool that automatically upgrades our database from our models.
Let’s get started!
I created a repo for the code in this project. Before getting started, make sure you have PostgreSQL running locally with a fresh new database. I use https://postgresapp.com/ for Mac.
git clone git@github.com:maciejzukowski/graphql-api-tutorial.git
cd graphql-api-tutorial
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txtCreate Database Models
We’re going to need a few tables:
- User —Store our users’ names, emails, and passwords
- Joke — Holds our jokes and which user created it
- UserLog — For small personal projects I like to store my logs in the database. It helps me debug issues, spot bad users, and monitor what’s happening on my system. This should not be used for large scale apps. It will take down your database.
#
# models.py
#import app.db
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from werkzeug.security import generate_password_hash, check_password_hash
import sqlalchemy as sa
import app.config
import jwt
import datetime
Base = declarative_base()
class User(Base):
__tablename__ = 'user'
id = sa.Column(sa.Integer, primary_key=True)
email = sa.Column(sa.String(120), index=True, unique=True)
name = sa.Column(sa.String)
password_hash = sa.Column(sa.String(128))
created_at = sa.Column(sa.DateTime, server_default=sa.sql.expression.text('NOW()'), nullable=False)
updated_at = sa.Column(sa.DateTime, server_default=sa.sql.expression.text('NOW()'), nullable=False)
def set_password(self, password):
self.password_hash = generate_password_hash(password)
def check_password(self, password):
return check_password_hash(self.password_hash, password)
def encode_auth_token(self):
try:
payload = {
'exp': datetime.datetime.utcnow() + datetime.timedelta(days=180, seconds=0),
'iat': datetime.datetime.utcnow(),
'sub': self.id
}
return jwt.encode(
payload,
app.config.get('SECRET_KEY'),
algorithm='HS256'
)
except Exception as e:
return e
@staticmethod
def decode_auth_token(request):
auth_header = request.headers.get('Authorization')
if auth_header:
auth_token = auth_header.split(" ")[1]
else:
auth_token = ''
try:
payload = jwt.decode(auth_token, app.config.get('SECRET_KEY'))
return payload['sub'], payload['admin']
except jwt.ExpiredSignatureError:
return None, None
# return 'Signature expired. Please log in again.'
except jwt.InvalidTokenError:
return None, None
except:
return None, None
# return 'Invalid token. Please log in again.'
class UserLog(Base):
__tablename__ = 'user_log'
id = sa.Column(sa.Integer, primary_key=True)
user_id = sa.Column(sa.ForeignKey("user.id"))
query = sa.Column(sa.Text)
variables = sa.Column(sa.Text)
user_agent = sa.Column(sa.Text)
response = sa.Column(sa.JSON)
created_at = sa.Column(sa.DateTime, server_default=sa.sql.expression.text('NOW()'), nullable=False)
user = relationship("User")
class Joke(Base):
__tablename__ = 'joke'
id = sa.Column(sa.Integer, primary_key=True)
text = sa.Column(sa.String())
user_id = sa.Column(sa.Integer, index=True)
created_at = sa.Column(sa.DateTime, server_default=sa.sql.expression.text('NOW()'), nullable=False)
We’ll be using JWT tokens for authentication. A JWT token is just a securely hashed string, sent in the headers, and then decoded on the server to get the current user id. Most of the code in the User db model is related to creating and managing this token.
Now it’s time to create the database tables with a migration. Alembic makes this easy! We run an autogenerate command to check the current database state, compare it with the models, and create the migration file. Then we run the upgrade command to run the migration file and update the database.
export DATABASE_URL=postgres:///myappdb
alembic revision -m "inital revision" --autogenerate
alembic upgrade head # updates the databaseNote: Make sure you’re in the virtual environment by running`source venv/bin/activate` when opening a new terminal tab.
GraphQL Server setup
Our server really only has one endpoint and it’s /graphql. All we need to do is setup our Flask app with a few configurations and add the graphql endpoint. The LoggingMiddleware and after_request logic deals with storing every request and response in the UserLog table. Trust me this is awesome and will save your a**. Did you get an email from an angry user about missing data? Check the UserLog table, to possibly find out out they deleted their own data, and let them know accordingly. Once again, for apps at scale, you wouldn’t put logs in your database. It would be fine for small projects.
#
# server.py
#import json
import os
from app.db import db_session
from app.models import User, UserLog
from flask import Flask, request, g
from app import db
from flask_graphql import GraphQLView
from app.graphql.schema import schema
from flask_cors import CORS
app = Flask(__name__)
app.debug = os.environ.get('ENV') != 'production'
app.config['SEND_FILE_MAX_AGE_DEFAULT'] = 0
app.secret_key = os.environ.get("SECRET_KEY")
cors = CORS(app)
app.config['CORS_HEADERS'] = 'Content-Type'
class LoggingMiddleware(object):
def resolve(self, next, root, info, **kwargs):
if root is None:
# Never store raw passwords anywhere
ignore = ['signUp','login','IntrospectionQuery']
user_id = User.decode_auth_token(request)
if info.operation.name.value not in ignore:
with db.session() as session:
log = UserLog(
user_id=user_id,
query=json.dumps(request.json['query']),
variables=json.dumps(request.json['variables']),
user_agent=request.headers.environ.get('HTTP_USER_AGENT')+"---"+request.headers.environ.get('REMOTE_ADDR')
)
session.add(log)
session.commit()
if info.operation.name.value:
g.log_id = log.id
return next(root, info, **kwargs)
app.add_url_rule(
"/graphql", view_func=GraphQLView.as_view(
"graphql",
schema=schema,
graphiql=os.environ.get('ENV') != 'production',
get_context=lambda:{'session': db_session},
middleware=[LoggingMiddleware()]
)
)
@app.after_request
def after_request(response):
if g.get('log_id', False):
if response.is_json:
with db.session() as session:
session.query(UserLog). \
filter(UserLog.id == g.log_id). \
update({"response": response.json})
return response
@app.teardown_appcontext
def shutdown_session(context):
db_session.remove()
if __name__ == "__main__":
app.run(debug=True, port=os.environ.get("PORT"), host=os.environ.get("HOST"))
All about dat GraphQL
Ok, now it’s time for the meat and potatoes: building the graphQL schema. We need to define all the queries, mutations, and types. I’ve seen a lot of different folder structures for setting up the schema. For simple apps, the structure I laid out works well for me. Once resolvers start growing in logic, you should start thinking about breaking them up into smaller files.
For those of you coming from REST and have yet to use GraphQL, I suggest first reading up on the topic at https://graphql.org/learn/. This will help you get to know the terminology and ideas.
# My File Structure
app/graphql
-[model_name]
--types.py #Defines your types and connects them to db model_name
--queries.py #Pretty much the GET requests for model_name
--mutations.py #The POST & UPDATE requests-schema.py #Runs through the folders & files and builds the schema
-base.py #Base classes for types, queries, and mutations which you need to subclass for them to be discovered.
Let’s run through creating the types, queries, and mutations for the Joke model.
#
# app/graphql/joke/types.py
#from graphene_sqlalchemy import SQLAlchemyObjectType
from app.graphql.base import BaseType
import app.models as m
class Joke(BaseType, SQLAlchemyObjectType):
class Meta:
model = m.Joke
We’re gonna be creating, editing, and returning Jokes, so we need to describe what a Joke is to GraphQL. As we already did this for our db model, SQLAlchemyObjectType will do most of the hard work pulling the fields from there. All we have to do is set the correct reference in the meta class and the Joke type will be ready to be queried and mutated.
Now it’s time to be able to request all the jokes or just one by ID. If we compare it to REST, it would be similar to GET /jokes and GET /jokes/1.
#
# app/graphql/joke/queries.py
#import graphene
import app.models as models
from app.graphql.base import BaseQuery
from .types import Joke
class Query(BaseQuery):
jokes = graphene.List(Joke)
joke = graphene.Field(Joke, id=graphene.ID())
def resolve_jokes(self, info):
session = info.context['session']
jokes = session.query(models.Joke) \
.all()
return jokes
def resolve_joke(self, info, id):
session = info.context['session']
return session.query(models.Joke) \
.filter(
models.Joke.id==id
) \
.one()
Here we define the two things we can query: jokes, which returns an array of jokes, and joke which returns just a joke but takes an id as an argument. You can see that the method being invoked takes the format resolve_[query_name] and that the arguments are being passed to these functions.
To connect to our database, we must grab the session from our context which we set for every request in server.py.
Tip — I use GraphiQL electron app to test and build my graphql endpoints
Ok, time to spin up the server and test out our graphql endpoints.
export SECRET_KEY=XXXXXXX #Create a secret key!
export DATABASE_URL=postgres:///myapp
PORT=3001 python app/server.pyIf you dive into GraphiQL and explore the Query schema, you will be able to see our new jokes and joke fields. If we run a quick query to test the jokes endpoint, we can correctly get back an empty array. However, there are no jokes in the system yet, let’s get to that.
Mutations take a slightly different form. Each one will be its own class, and the logic will happen in the mutate function. We will use the authentication_required decorator to make sure the user is logged in to perform the action.
#
# app/graphql/joke/mutations.py
#import graphene
from graphql import GraphQLError
from app.graphql.base import BaseMutation
from app.helpers.security import authentication_required
from .types import Joke
import app.models as models
class CreateJoke(BaseMutation):
class Arguments:
text = graphene.String()
ok = graphene.Boolean()
joke = graphene.Field(lambda: Joke)
@authentication_required
def mutate(self, info, text):
session = info.context['session']
current_user_id = info.context['current_user_id']
joke = models.Joke(
text=text,
user_id=current_user_id
)
session.add(joke)
session.commit()
return CreateJoke(joke=joke, ok=True)
class EditJoke(BaseMutation):
class Arguments:
id = graphene.ID()
text = graphene.String()
ok = graphene.Boolean()
joke = graphene.Field(lambda: Joke)
@authentication_required
def mutate(self, info, id, text):
session = info.context['session']
current_user_id = info.context['current_user_id']
joke = session.query(models.Joke) \
.filter(
models.Joke.id == id,
models.Joke.user_id==current_user_id
) \
.one_or_none()
if not joke:
raise GraphQLError('Not authorized')
joke.text=text
session.commit()
return EditJoke(joke=joke, ok=True)
The repo has all the graphql queries, mutations, and types for User to login, sign up, and change passwords. The process is very similar to what we just did for Joke, therefore I will not be going over that in this tutorial. Check the source code at graphql/user/ to understand the details. Let’s create a user in the DB!
The signUp mutation returns a token that we’ll need to put in the headers if we want to do authenticated calls.
Let’s create a joke! The response will return the newly created joke.
Now, let’s check the jokes query to see if it shows up.
It does!
We now have a backend that allows anyone to see a list of all the jokes, sign up/login, and add their own joke for others to see! It’s pretty much reddit.
We also log requests into the user_log table. I love Postico to connect to my production and local PostgreSQL databases. The screenshot above is just a quick glance at what the logs look like.
Well that’s it for the backend! Checkout the github repo here. Feel free to leave any feedback or comments. Stay tuned for the React Native app tutorial!
