In this day and age, if you've worked on anything internet related, you have probably come across the word API. This stands for Application Programming Interface. This basically defines the interactions between multiple software interfaces, be it online or offline. Now what is the first thing that comes to mind when you think of online APIs? REST right?

Well, GraphQL is here to change that.

What is GraphQL?

As defined on their website:

GraphQL is a query language for APIs and a runtime for fulfilling those queries with your existing data. GraphQL provides a complete and understandable description of the data in your API, gives clients the power to ask for exactly what they need and nothing more, makes it easier to evolve APIs over time, and enables powerful developer tools.

It was created by Facebook in 2012 and used by them for years before coming out as open-source in 2015, and is used by multiple companies, such as Github, Coursera, Pinterest amongst many others.

Why Use GraphQL?

It makes querying for data infinitely easier, by availing all the data a client would need without having to make many different endpoints as we would have to using REST. It also comes with it's own native editor, that can allow you to see what possible issues you may have with your queries called graphiQL, allowing you to test your queries quickly and efficiently.

At the same time, with its native typing, it allows a client querying to know exactly what they are supposed to send, and if they do not, it offers very helpful and insightful tips as to how to fix it.

Now that we've gotten that out of the way, let us find out how to implement a simple API using FastAPI, SQLAlchemy and MySQL. All you need for this is base level knowledge in Python, and a tiny bit of MySQL.

Let's begin.

Installation

First things first, you need to install FastAPI. FastAPI is a relatively new API framework that was created by Sebastian Ramirez that he calls "starlette on steroids". Basically it takes the best parts of Starlette and Flask and groups them into a wonderful framework.

To install it, simply go to your command line and input:

pip install fastapi

to use Python's native pip package installer to install it. Once it has been installed, follow it up by installing sqlalchemy to allow us to interact with the database.

pip install sqlalchemy

sqlalchemy is an ORM (Object-relational mapper) that abstracts the database and makes it easier to interact with, basically making writing queries less of a chore.

Starting the Project

We're going to be creating a simple API that allows a user to add a campaign, all people to contribute to it, and monitor how it has progressed.

First things first, create a folder called 'lipiame' (the name of the product) with the following structure.

├── lipiame
│   ├──models.py
│   ├──schema.py
|   ├──settings.py

models.py is where the sqlalchemy models are going to be,                             schema.py is where the graphql queries and mutations (more on these later) are going to live and settings.py is where the sqlalchemy connection is going to lie, among other things. Let's start with the settings.py file.

from sqlalchemy import create_engine

engine = create_engine('mysql+pymysql://root:@localhost/lipiame', pool_recycle=3600)

This is what we use to connect to the database.

The database binding uses pymysql, so if you do not have it, you can install it using pip:

pip install pymysql

The rest of the string is broken down like this: 'username:password@host/database'. In our case, seeing as it is running locally and therefore does not have a password, we leave that part blank.

This is how models.py looks:

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, DateTime, Text, Date
from settings import engine
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Campaign(Base):
    __tablename__ = 'campaign'
    id = Column(Integer, primary_key=True, autoincrement=True)
    campaign_name = Column(String(50))
    amount_contributed = Column(Integer, default=0)
    user_email = Column(String(50))
 

Base.metadata.create_all(engine)

What this does is create a table called 'campaign', with the campaign name, the amount that has been contributed so far (which we set as the default of 0) and the email it is associated to.

At the bottom of the code block there is:

Base.metadata.create_all(engine)

which creates the tables in the database. We call engine from the settings file so we can actually access the table. Make sure the database exists before running this otherwise it will throw an error. Upon running this on your command line:

python models.py

the tables should be created. Now let's go to the heart of the project, in schema.py.

import graphene
from starlette.applications import Starlette
from starlette.routing import Route
from starlette.graphql import GraphQLApp
from models import Campaign
from settings import engine
from sqlalchemy.orm import sessionmaker, scoped_session
from graphene_sqlalchemy import SQLAlchemyObjectType, SQLAlchemyConnectionField



Session = scoped_session(sessionmaker(bind=engine))


class CreateCampaign(graphene.Mutation):
    id = graphene.Int()
    campaign_name = graphene.String()
    user_email = graphene.String() 


    class Arguments:
        campaign_name = graphene.String()
        user_email = graphene.String()

    def mutate(self, info, user_email, campaign_name):
        session = Session()
        campaign = Campaign(user_email=user_email, campaign_name=campaign_name)
        session.add(campaign)
        session.commit()

        return CreateCampaign(
            id=campaign.id,
            campaign_name=campaign.campaign_name,
            user_email=campaign.user_email,
        )


class Mutation(graphene.ObjectType):
    create_campaign = CreateCampaign.Field()


class CampaignType(SQLAlchemyObjectType):
    class Meta:
        model = Campaign	


class Query(graphene.ObjectType):
    campaign = graphene.List(CampaignType,
    	                     user_email=graphene.String())

    def resolve_campaign(self, info, user_email):
        session = Session()
        campaign = session.query(Campaign).filter(Campaign.user_email == user_email).all()
        return campaign


routes = [
    Route('/', GraphQLApp(schema=graphene.Schema(query=Query, mutation=Mutation)))
]

app = Starlette(routes=routes)

It's a lot, but let's break it down.

import graphene
from starlette.applications import Starlette
from starlette.routing import Route
from starlette.graphql import GraphQLApp
from models import Campaign
from settings import engine
from sqlalchemy.orm import sessionmaker, scoped_session
from graphene_sqlalchemy import SQLAlchemyObjectType, SQLAlchemyConnectionField

FastAPI graphQL runs on Starlette, hence all the imports from it, the Route sets which URL graphiQL is going to run on, and GraphQLApp is graphiQL itself.

Graphene and graphene_sqlalchemy are necessary for the bindings to the tables in the database.

We import the Campaign class from models that was the definition of the table.

We import the engine, which is used to connect to the database.

Session = scoped_session(sessionmaker(bind=engine))

This simply creates the session to enable threading safely so there won't be overlaps with variables from the database. Now before we look at our first mutation, let's define what it is.

A mutation is basically anything that modifies data in a database. It's that simple.

Now let's see how the first one looks.

class CreateCampaign(graphene.Mutation):
    id = graphene.Int()
    campaign_name = graphene.String()
    user_email = graphene.String() 

These values represent what is to be returned once the values are added to the table, so in this case, it will return the record id, the name, and the user email associated to.

    class Arguments:
        campaign_name = graphene.String()
        user_email = graphene.String()

This defines what the mutation accepts. In this case, the campaign name, and the user email.

    def mutate(self, info, user_email, campaign_name):

This is where the variables are passed. Info is used in relays, but we won't use it in this case.

        session = Session()

This is where the database connection is opened.

        campaign = Campaign(user_email=user_email, campaign_name=campaign_name)
        session.add(campaign)
        session.commit()

Now this is the meat of the mutation. We use the Campaign model we created, and add the user_email and campaign_name that have been passed. We then add it to the session, and then commit it.

        return CreateCampaign(
            id=campaign.id,
            campaign_name=campaign.campaign_name,
            user_email=campaign.user_email,
        )

This is what is returned back to the front end, the id, campaign_name and user_email.

class Mutation(graphene.ObjectType):
    create_campaign = CreateCampaign.Field()

This is what defines the mutations that can be availed to the clients. If your created mutation is not put in this class, it will not be accessible.

Now let's break down the query.

class CampaignType(SQLAlchemyObjectType):
    class Meta:
        model = Campaign	

This class is what allows GraphQL to be able to access the Campaign table.

class Query(graphene.ObjectType):
    campaign = graphene.List(CampaignType,
    	                     user_email=graphene.String())

The class Query is where all queries are put. In this case, we call the query 'campaign', and pass the CampaignType class that we defined above for the query to work. We then define the input we expect to get from the client, in this case, user_email.

    def resolve_campaign(self, info, user_email):
        session = Session()
        campaign = session.query(Campaign).filter(Campaign.user_email == user_email).all()
        return campaign

The query is called a 'resolver' hence why it is set up as 'resolve_campaign' here. We pass the user_email we get, and use that to query the table using sqlalchemy, specifically the user_email table, after which we return all the results that match the query.

routes = [
    Route('/', GraphQLApp(schema=graphene.Schema(query=Query, mutation=Mutation)))
]

app = Starlette(routes=routes)

With this, we define endpoint where we want our graphiQL editor. We then pass our Query and Mutation classes through the schema for them to available, and then run it via the app.

Now let's run this and see it in action.

First things first, install Uvicorn using pip.

pip install uvicorn

This is what we use to be able to run the code. Navigate to the folder with the schema.py and run it:

uvicorn schema:app --host 127.0.0.1 --port 8000

Now if you go to http://127.0.0.1:8000/, you should see the following:

And you can start writing graphQL queries! The best part about graphiQL, the autocomplete.

This makes it very simple to write the queries and check if everything is okay.

Let's create a record.

And it's as simple as that. Within the curly braces, we can decide what we want returned. If we only want the record id, we can just pass id. But! You cannot query anything more than what we set earlier, specifically, the record id, the email and campaign name.

One thing to note is that, even if you've saved the column names in snake case, graphQL overwrites it and sets them in camel case, but only in the queries, everywhere else they maintain their snake case properties.

Now let's query that record to see how much has been contributed.

Nothing yet :(

So let's write something to enable someone to contribute.

class CampaignContribute(graphene.Mutation):
    id = graphene.Int()
    campaign_name = graphene.String()
    amount_contributed = graphene.Int() 


    class Arguments:
        campaign_id = graphene.String()
        contribution = graphene.Int()

    def mutate(self, info, campaign_id, contribution):
        session = Session()
        campaign = session.query(Campaign).filter(Campaign.id == campaign_id).first()
        campaign.amount_contributed += contribution
        session.add(campaign)

        return CampaignContribute(
            id=campaign.id,
            campaign_name=campaign.campaign_name,
            amount_contributed=campaign.amount_contributed,
        )

The difference between this and the previous mutation is that this one gets a record that already exists and updates it, in this case, the amount contributed. This record is queried using the record id.

Now someone is feeling generous and wants to contribute, so let's avail this and allow them to contribute.

class Mutation(graphene.ObjectType):
    create_campaign = CreateCampaign.Field()
    campaign_contribute = CampaignContribute.Field()

And now we contribute.

Someone wants me to go to university, yay!

Conclusion

This was a small taste of how graphQL works with FastAPI and sqlalchemy. You see how simple it can be, accessing a database and interacting with it. Now go out there and was the taste of REST out of your mouth with this new and fun experience.

This project can be found on my here

You've successfully subscribed to Decoded For Devs
Welcome back! You've successfully signed in.
Great! You've successfully signed up.
Your link has expired
Success! Your account is fully activated, you now have access to all content.