Flask Graphene SQLAlchemy Tutorial - alexisrolland/flask-graphene-sqlalchemy GitHub Wiki

The objective of this tutorial is to describe step by step how to create a GraphQL API using Python and more specifically the following packages:

The API will expose data coming from an SQLite database stored locally on your machine. The end result will follow the files architecture described in the schema below. This files architecture has been reworked compared to previous tutorials I have followed on the topic with the intention to make it more scalable. It also contains complete examples of GraphQL mutations which I was not able to find in existing tutorials:

example/
├── api.py
├── schema.py
├── schema_people.py
├── schema_planet.py
├── setup.py
├── utils.py
└── database/
    ├── base.py
    ├── database.db
    ├── model_people.py
    ├── model_planet.py
    └── data/
        ├── people.json
        └── planet.json

Architecture

  • The example folder contains the files used by Flask to start the API and the schema files used by Graphene to define GraphQL queries and mutations. It also contains the database sub folder.
  • The database sub folder contains files used to create and interact with the SQLite database. It contains in particular one model file per database table which maps it to the corresponding SQLAlchemy class.

Project setup

Open a terminal window to create your Python virtual environment. Keep the environment name short for convenience. I named it gql in the example below.

$ sudo apt-get install python3-venv
$ python3 -m venv /mypath/gql
$ source /mypath/gql/bin/activate

Remark: The last command above is used to activate your virtual environment. You can deactivate it by typing deactivate in your terminal.

Make sure your virtual environment is activated and install the following third party packages. I have specifically indicated the versions I used for this tutorial.

$ pip3 install sqlalchemy==1.1.14
$ pip3 install graphene==2.0.0
$ pip3 install graphene-sqlalchemy==2.0.0
$ pip3 install flask==0.11.1
$ pip3 install flask-graphql==1.4.1

Create your project folders and change directory to it.

$ mkdir example
$ mkdir example/database
$ cd example/

Create database

The database will contain some Planet and People data from Star Wars stored in 2 different tables (data source: The Star Wars API):

  • Planet: List of planets appearing in Star Wars movies
  • People: List of people appearing in Star Wars movies

There is a relationship between planets and people as one planet is the homeworld of one or several persons. It is a "one-to-many" relationship (one planet, many people). The planet_id column in the people table will be used as a foreign key of the planet table.

Data model

In the database sub folder create a file named base.py. This file will be used by SQLALchemy to perform the following:

  • Define the database engine used to generate and interact with the SQLite file database.db.
  • Create the Base class used to produce database tables and map them to their respective SQLAlchemy class.
  • Manage the database sessions in order to execute queries against the database.

base.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session, sessionmaker
import os

# Create database engine
db_name = 'database.db'
db_path = os.path.join(os.path.dirname(__file__), db_name)
db_uri = 'sqlite:///{}'.format(db_path)
engine = create_engine(db_uri, convert_unicode=True)

# Declarative base model to create database tables and classes
Base = declarative_base()
Base.metadata.bind = engine  # Bind engine to metadata of the base class

# Create database session object
db_session = scoped_session(sessionmaker(bind=engine, expire_on_commit=False))
Base.query = db_session.query_property()  # Used by graphql to execute queries

In the database sub folder create two files named model_people.py and model_planet.py which will contain the data model configuration required by SQLAlchemy to create the tables and their relationship.

model_people.py

from .base import Base
from sqlalchemy import Column, ForeignKey, Integer, String


class ModelPeople(Base):
    """People model."""

    __tablename__ = 'people'

    id = Column('id', Integer, primary_key=True)
    name = Column('name', String)
    height = Column('height', String)
    mass = Column('mass', String)
    hair_color = Column('hair_color', String)
    skin_color = Column('skin_color', String)
    eye_color = Column('eye_color', String)
    birth_year = Column('birth_year', String)
    gender = Column('gender', String)
    planet_id = Column('planet_id', Integer, ForeignKey('planet.id'))
    created = Column('created', String)
    edited = Column('edited', String)
    url = Column('url', String)

model_planet.py

from .base import Base
from .model_people import ModelPeople
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import relationship


class ModelPlanet(Base):
    """Planet model."""

    __tablename__ = 'planet'

    id = Column('id', Integer, primary_key=True)
    name = Column('name', String)
    rotation_period = Column('rotation_period', String)
    orbital_period = Column('orbital_period', String)
    diameter = Column('diameter', String)
    climate = Column('climate', String)
    gravity = Column('gravity', String)
    terrain = Column('terrain', String)
    surface_water = Column('surface_water', String)
    population = Column('population', String)
    created = Column('created', String)
    edited = Column('edited', String)
    url = Column('url', String)

    peopleList = relationship(ModelPeople, backref='planet')

Remark: The import of the class ModelPeople is required in the file model_planet in order to properly create the relationship between ModelPeople and ModelPlanet. This is in particular necessary for Graphene to be able to retrieve people when querying on planets and vice versa.

The next file setup.py will create the database database.db and load data into it. Data is available in the project repository in the shape of JSON files:

In the database folder create another data sub folder and place the JSON files in it. Your path should be something like:

  • example/database/data/people.json
  • example/database/data/planet.json

Copy the following code and create the setup.py file in the example folder.

setup.py

from ast import literal_eval
from database.model_people import ModelPeople
from database.model_planet import ModelPlanet
from database import base
import logging
import sys

# Load logging configuration
log = logging.getLogger(__name__)
logging.basicConfig(
    stream=sys.stdout,
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')


if __name__ == '__main__':
    log.info('Create database {}'.format(base.db_name))
    base.Base.metadata.create_all(base.engine)

    log.info('Insert Planet data in database')
    with open('database/data/planet.json', 'r') as file:
        data = literal_eval(file.read())
        for record in data:
            planet = ModelPlanet(**record)
            base.db_session.add(planet)
        base.db_session.commit()

    log.info('Insert People data in database')
    with open('database/data/people.json', 'r') as file:
        data = literal_eval(file.read())
        for record in data:
            planet = ModelPeople(**record)
            base.db_session.add(planet)
        base.db_session.commit()

Remark: It is important to import ModelPeople before ModelPlanet because the relationship between the two classes has been defined in the ModelPlanet class. This means SQLAlchemy requires ModelPeople to be created first in order to be able to create ModelPlanet.

Execute setup.py. You should see a new file database.db appearing in the database folder.

$ python3 setup.py
2018-02-03 15:36:37,561 - __main__ - INFO - Create database database.db
2018-02-03 15:36:37,589 - __main__ - INFO - Insert Planet data in database
2018-02-03 15:36:37,610 - __main__ - INFO - Insert People data in database

Create GraphQL queries

The database has been created and contains our data. We can start to create the API. In the example folder create two files named schema_people.py and schema_planet.py. These two files use the Graphene-SQLAlchemy package to map our ModelPeople and ModelPlanet classes to Graphene schema objects. This allows us to reuse SQLAlchemy classes definition instead of redefining Graphene schemas manually.

schema_people.py

from graphene_sqlalchemy import SQLAlchemyObjectType
from database.model_people import ModelPeople
import graphene


# Create a generic class to mutualize description of people attributes for both queries and mutations
class PeopleAttribute:
    name = graphene.String(description="Name of the person.")
    height = graphene.String(description="Height of the person.")
    mass = graphene.String(description="Mass of the person.")
    hair_color = graphene.String(description="Hair color of the person.")
    skin_color = graphene.String(description="Skin color of the person.")
    eye_color = graphene.String(description="Eye color of the person.")
    birth_year = graphene.String(description="Birth year of the person.")
    gender = graphene.String(description="Gender of the person.")
    planet_id = graphene.ID(description="Global Id of the planet from which the person comes from.")
    url = graphene.String(description="URL of the person in the Star Wars API.")


class People(SQLAlchemyObjectType, PeopleAttribute):
    """People node."""

    class Meta:
        model = ModelPeople
        interfaces = (graphene.relay.Node,)

schema_planet.py

from graphene_sqlalchemy import SQLAlchemyObjectType
from database.model_planet import ModelPlanet
import graphene


# Create a generic class to mutualize description of planet attributes for both queries and mutations
class PlanetAttribute:
    name = graphene.String(description="Name of the planet.")
    rotation_period = graphene.String(description="Rotation period of the planet.")
    orbital_period = graphene.String(description="Orbital period of the planet.")
    diameter = graphene.String(description="Diameter of the planet.")
    climate = graphene.String(description="Climate period of the planet.")
    gravity = graphene.String(description="Gravity of the planet.")
    terrain = graphene.String(description="Terrain of the planet.")
    surface_water = graphene.String(description="Surface water of the planet.")
    population = graphene.String(description="Population of the planet.")
    url = graphene.String(description="URL of the planet in the Star Wars API.")


class Planet(SQLAlchemyObjectType, PlanetAttribute):
    """Planet node."""

    class Meta:
        model = ModelPlanet
        interfaces = (graphene.relay.Node,)

In the example folder create a file named schema.py. It is used to declare the different objects on which the API can perform queries. It will contain 4 objects for the moment:

  • people: to perform a query that returns a single person
  • peopleList: to perform a query that returns a list of persons
  • planet: to perform a query that returns a single planet
  • planetList: to perform a query that returns a list of planets

schema.py

from graphene_sqlalchemy import SQLAlchemyConnectionField
import graphene
import schema_planet
import schema_people


class Query(graphene.ObjectType):
    """Query objects for GraphQL API."""

    node = graphene.relay.Node.Field()
    people = graphene.relay.Node.Field(schema_people.People)
    peopleList = SQLAlchemyConnectionField(schema_people.People)
    planet = graphene.relay.Node.Field(schema_planet.Planet)
    planetList = SQLAlchemyConnectionField(schema_planet.Planet)


schema = graphene.Schema(query=Query)

Create Flask application

In the example folder create a file api.py with the following code. This file is the main entry point of our application. It creates the Flask application, specifies the URL of our API and terminates the database sessions after each http requests.

api.py

from database.base import db_session
from flask import Flask
from flask_graphql import GraphQLView
from schema import schema

app = Flask(__name__)
app.add_url_rule(
    '/graphql',
    view_func=GraphQLView.as_view('graphql', schema=schema, graphiql=True))


@app.teardown_appcontext
def shutdown_session(exception=None):
    db_session.remove()


if __name__ == '__main__':
    app.run(threaded=True, debug=True)

Start your api by typing the following command in a terminal window.

python3 api.py

Queries examples

Open your browser and navigate to your API GraphiQL interface: http://127.0.0.1:5000/graphql

GraphiQL

Get list of people: This query returns a list of people and their planet.

query {
  peopleList {
    edges {
      node {
        id
        name
        height
        mass
        hairColor
        skinColor
        eyeColor
        birthYear
        gender
        created
        edited
        url
        planet {
          id
          name
          rotationPeriod
          diameter
          climate
          gravity
          terrain
          surfaceWater
          population
          created
          edited
          url
        }
      }
    }
  }
}

Get one planet: This query returns one planet and its people.

query {
  planet (id: "UGxhbmV0OjE="){
    id
    name
    rotationPeriod
    diameter
    climate
    gravity
    terrain
    surfaceWater
    population
    created
    edited
    url
    peopleList {
      edges {
        node {
          id
          name
          height
          mass
          hairColor
          skinColor
          eyeColor
          birthYear
          gender
          created
          edited
          url
        }
      }
    }
  }
}

Add GraphQL mutations

In the example folder create a file utils.py which will contain utility methods used by Graphene mutations. It includes in particular a method to convert Graphene inputs sent in the http request to a dictionary object. This is necessary for SQLAlchemy to to be able to use the inputs to insert or update database records. This method also converts GraphQL global Ids into their corresponding database Ids so that SQLAlchemy can use them to query the database.

utils.py

from graphql_relay.node.node import from_global_id


def input_to_dictionary(input):
    """Method to convert Graphene inputs into dictionary"""
    dictionary = {}
    for key in input:
        # Convert GraphQL global id to database id
        if key[-2:] == 'id':
            input[key] = from_global_id(input[key])[1]
        dictionary[key] = input[key]
    return dictionary

Update the file schema_people.py as below to add classes for the createPerson and updatePerson mutations.

schema_people.py

from datetime import datetime
from graphene_sqlalchemy import SQLAlchemyObjectType
from database.base import db_session
from database.model_people import ModelPeople
import graphene
import utils


# Create a generic class to mutualize description of people attributes for both queries and mutations
class PeopleAttribute:
    name = graphene.String(description="Name of the person.")
    height = graphene.String(description="Height of the person.")
    mass = graphene.String(description="Mass of the person.")
    hair_color = graphene.String(description="Hair color of the person.")
    skin_color = graphene.String(description="Skin color of the person.")
    eye_color = graphene.String(description="Eye color of the person.")
    birth_year = graphene.String(description="Birth year of the person.")
    gender = graphene.String(description="Gender of the person.")
    planet_id = graphene.ID(description="Global Id of the planet from which the person comes from.")
    url = graphene.String(description="URL of the person in the Star Wars API.")


class People(SQLAlchemyObjectType):
    """People node."""

    class Meta:
        model = ModelPeople
        interfaces = (graphene.relay.Node,)


class CreatePersonInput(graphene.InputObjectType, PeopleAttribute):
    """Arguments to create a person."""
    pass


class CreatePerson(graphene.Mutation):
    """Mutation to create a person."""
    person = graphene.Field(lambda: People, description="Person created by this mutation.")

    class Arguments:
        input = CreatePersonInput(required=True)

    def mutate(self, info, input):
        data = utils.input_to_dictionary(input)
        data['created'] = datetime.utcnow()
        data['edited'] = datetime.utcnow()

        person = ModelPeople(**data)
        db_session.add(person)
        db_session.commit()

        return CreatePerson(person=person)


class UpdatePersonInput(graphene.InputObjectType, PeopleAttribute):
    """Arguments to update a person."""
    id = graphene.ID(required=True, description="Global Id of the person.")


class UpdatePerson(graphene.Mutation):
    """Update a person."""
    person = graphene.Field(lambda: People, description="Person updated by this mutation.")

    class Arguments:
        input = UpdatePersonInput(required=True)

    def mutate(self, info, input):
        data = utils.input_to_dictionary(input)
        data['edited'] = datetime.utcnow()

        person = db_session.query(ModelPeople).filter_by(id=data['id'])
        person.update(data)
        db_session.commit()
        person = db_session.query(ModelPeople).filter_by(id=data['id']).first()

        return UpdatePerson(person=person)

Update the file schema_planet.py as below to add classes for the createPlanet and updatePlanet mutations.

schema_planet.py

from datetime import datetime
from graphene_sqlalchemy import SQLAlchemyObjectType
from database.base import db_session
from database.model_planet import ModelPlanet
import graphene
import utils


# Create a generic class to mutualize description of planet attributes for both queries and mutations
class PlanetAttribute:
    name = graphene.String(description="Name of the planet.")
    rotation_period = graphene.String(description="Rotation period of the planet.")
    orbital_period = graphene.String(description="Orbital period of the planet.")
    diameter = graphene.String(description="Diameter of the planet.")
    climate = graphene.String(description="Climate period of the planet.")
    gravity = graphene.String(description="Gravity of the planet.")
    terrain = graphene.String(description="Terrain of the planet.")
    surface_water = graphene.String(description="Surface water of the planet.")
    population = graphene.String(description="Population of the planet.")
    url = graphene.String(description="URL of the planet in the Star Wars API.")


class Planet(SQLAlchemyObjectType):
    """Planet node."""

    class Meta:
        model = ModelPlanet
        interfaces = (graphene.relay.Node,)


class CreatePlanetInput(graphene.InputObjectType, PlanetAttribute):
    """Arguments to create a planet."""
    pass


class CreatePlanet(graphene.Mutation):
    """Create a planet."""
    planet = graphene.Field(lambda: Planet, description="Planet created by this mutation.")

    class Arguments:
        input = CreatePlanetInput(required=True)

    def mutate(self, info, input):
        data = utils.input_to_dictionary(input)
        data['created'] = datetime.utcnow()
        data['edited'] = datetime.utcnow()

        planet = ModelPlanet(**data)
        db_session.add(planet)
        db_session.commit()

        return CreatePlanet(planet=planet)


class UpdatePlanetInput(graphene.InputObjectType, PlanetAttribute):
    """Arguments to update a planet."""
    id = graphene.ID(required=True, description="Global Id of the planet.")


class UpdatePlanet(graphene.Mutation):
    """Update a planet."""
    planet = graphene.Field(lambda: Planet, description="Planet updated by this mutation.")

    class Arguments:
        input = UpdatePlanetInput(required=True)

    def mutate(self, info, input):
        data = utils.input_to_dictionary(input)
        data['edited'] = datetime.utcnow()

        planet = db_session.query(ModelPlanet).filter_by(id=data['id'])
        planet.update(data)
        db_session.commit()
        planet = db_session.query(ModelPlanet).filter_by(id=data['id']).first()

        return UpdatePlanet(planet=planet)

Update the file schema.py to add the mutation class.

schema.py

from graphene_sqlalchemy import SQLAlchemyConnectionField
import graphene
import schema_planet
import schema_people


class Query(graphene.ObjectType):
    """Query objects for GraphQL API."""

    node = graphene.relay.Node.Field()
    people = graphene.relay.Node.Field(schema_people.People)
    peopleList = SQLAlchemyConnectionField(schema_people.People)
    planet = graphene.relay.Node.Field(schema_planet.Planet)
    planetList = SQLAlchemyConnectionField(schema_planet.Planet)


class Mutation(graphene.ObjectType):
    createPerson = schema_people.CreatePerson.Field()
    updatePerson = schema_people.UpdatePerson.Field()
    createPlanet = schema_planet.CreatePlanet.Field()
    updatePlanet = schema_planet.UpdatePlanet.Field()


schema = graphene.Schema(query=Query, mutation=Mutation)

Mutations examples

Create a person: This mutation creates a new person in the people table and returns its corresponding data.

mutation {
  createPerson (input: {
    name: "Alexis ROLLAND"
    height: "189"
    mass: "73"
    hairColor: "brown"
    skinColor: "white"
    eyeColor: "green"
    birthYear: "1984"
    gender: "male"
    planetId: "UGxhbmV0Ojk="
  }) {
    person {
      id
      name
      height
      mass
      hairColor
      skinColor
      eyeColor
      birthYear
      gender
      planet {
        id
        name
      }
      created
      edited
      url
    }
  }
}

Update a planet: This mutation updates a planet in the planet table and returns its corresponding data.

mutation {
  updatePlanet (input:{
    id: "UGxhbmV0Ojk="
    population: "1000000000001"
  }) {
    planet {
      id
      name
      rotationPeriod
      orbitalPeriod
      diameter
      climate
      terrain
      surfaceWater
      population
      created
      edited
      url
    }
  }
}

Voilà!