2022年5月28日土曜日

Building a GraphQL server (Relay support) with FastAPI, Graphene, and SQLAlchemy.The migration from REST API to GraphQL is underway.

https://qiita.com/woods0918/items/2d7b2421d8a8f1b6d724

Python

,

sqlalchemy

,

GraphQL

,

Graphene

,

FastAPI

Day 15 of the VISITS Technologies Advent Calendar 2020 will be handled by @woods0918.


The backend of ideagram, a service for which I am a product manager, is in the process of migrating from REST API to GraphQL.


Through my first Graphql and GraphQL Playground, I had some knowledge of GraphQL and experience with client-side operations, but I had no experience implementing a GraphQL server, so I thought I would give it a shot to make the conversation with the backend engineers smoother.


This time, I will implement it in Python, which I am familiar with writing, but the language employed in ideagram is Ruby (WAF : Ruby on Rails).


The code is here.


Python and library versions

pyproject.toml

python = "^3.8"

fastapi = "^0.62.0"

uvicorn = "^0.13.0"

SQLAlchemy = "^1.3.20"

graphene = "^2.1.8"

python-dotenv = "^0.15.0"

mysqlclient = "^2.0.2"

graphene-sqlalchemy = "^2.3.0"

Directory structure

# Excerpt only files necessary for explanation

|--.env # Define parameters for database connection

|--app

  |--__init__.py

  |--.cruds # Define queries to database

    |--__init__.py

    |--seeds.py

    |--todos.py

    |--users.py

  |---database.py # Define the connection to the database

  |--main.py # FastAPI endpoints

  |--models # Define classes used by SQLAlchemy

    |--__init__.py

    |--todos.py

    |--users.py

  |--schema # Define Query, Mutation, Schema, etc. used in Graphene

    |--__init__.py

    |---schema.py

    |--todos.py

    |--users.py

  |--setting.py

database.py

database.py

import os

from contextlib import contextmanager

from sqlalchemy import create_engine, inspect

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy.orm import scoped_session, sessionmaker

from dotenv import load_dotenv


load_dotenv()


DB_USERNAME = os.getenv("DB_USERNAME")

DB_PASSWORD = os.getenv("DB_PASSWORD")

DB_HOSTNAME = os.getenv("DB_HOSTNAME")

DB_PORT = os.getenv("DB_PORT")

DB_NAME = os.getenv("DB_NAME")

DB_CHARSET = os.getenv("DB_CHARSET")


ENGINE = create_engine(

    f "mysql://{DB_USERNAME}:{DB_PASSWORD}@{DB_HOSTNAME}:{DB_PORT}/{DB_NAME}?charset={DB_CHARSET}"

)

SESSION = scoped_session(

    sessionmaker(autocommit=False, autoflush=False, bind=ENGINE, expire_on_commit=False)

)


class SelfBase(object):

    def to_dict(self):

        model = {}

        for column in self.__table__.columns:

            model[column.name] = str(getattr(self, column.name))

        return model


BASE = declarative_base(cls=SelfBase)

BASE.query = SESSION.query_property()


@contextmanager

def session_scope():

    session = SESSION()

    try:

        yield session

        session.commit()

    except. session.rollback()

        session.rollback()

        raise

    finally:

        session.close()

In database.py, the connection to the database is defined.

DB parameters are read from .env using dotenv's load_dotenv.


SESSION is generated using scoped_session and sessionmaker.

When scoped_session is used, the same session is returned even if Session() is executed many times.


The SelfBase class (to be created later) is the base class for converting classes used in SQLAlchemy to dictionary type.

The base class for sqlalchemy is generated by inheriting from the SelfBase class.


If we do BASE.query = SESSION.query_property()

SESSION.query(HogeClass).filter(HogeClass.col1 == "hoge").all()

to the

HogeClass.query.filter(HogeClass.col1 == "hoge").all()

can be written as follows, which is plainly convenient.


Finally, session_scope is based on the article on how to handle sessions in SQLAlchemy.


with session_scope() as session:

    # some DB processing

I thought it would be convenient to write "with session_scope() as session: # some DB processing" so that I don't have to write "commit" and "close" every time.


Models

In models, we define the classes to be used in SQLAlchemy.


users.py

from sqlalchemy import Column

from sqlalchemy.dialects.mysql import INTEGER as Integer

from sqlalchemy.dialects.mysql import TEXT as Text

from app.database import BASE


class User(BASE):

    __tablename__ = "users"


    id = Column(Integer(unsigned=True), primary_key=True, unique=True, autoincrement=True)

    name = Column(Text)


    def __init__(self, name: str):

        self.name = name

We create it by inheriting from the sqlalchemy base class generated earlier in database.py.


Define the actual table name of the DB in __tablename__.


id and name are the columns of the table, which will be defined by importing data types from sqlalchemy.

If you want to use MySQL's own functions, import them from sqlalchemy.dialects.mysql.


Finally, define the initialization process with __init__.


cruds

seeds.py

In seeds.py, CREATE the table and INSERT the seed data.


seeds.py

import sys

import pathlib

from datetime import datetime


current_dir = pathlib.Path(__file__).resolve().parent

sys.path.append( str(current_dir) + '/... /... /' )


from app.database import BASE, ENGINE, session_scope

from app.models.todos import Todo

from app.models.users import User


def generate_seed_data():

    BASE.metadata.create_all(ENGINE)

    users = [["Taro"], ["Jiro"], ["Hanako"]]

    todos = [["todos"]]

        [1, "title1", "description1", datetime.now()],

        [1, "title2", "description2", datetime.now()],

        [2, "title3", "description3", datetime.now()],

        [2, "title4", "description4", datetime.now()],

        [3, "title5", "description5", datetime.now()],

        [3, "title6", "description6", datetime.now()]

    ]


    with session_scope() as session:

        for user in users:

            session.add(User(user[0]))

        for todo in todos:

            session.add(Todo(

                user_id = todo[0],

                title = todo[1],

                description = todo[2],

                deadline = todo[3])

            ))


if __name__ == "__main__":

    generate_seed_data()

First, BASE.metadata.create_all(ENGINE) generates the table.

If there are no changes from the existing table, it skips creating it and executes a "good" CREATE, which is very useful.


It then generates the Seed data and performs an INSERT with session_scope() as session: and thereafter.


users.py

(to be discussed later), we define the query to be used when performing Query and Mutation.


users.py

from typing import Dict, Optional

from app.database import session_scope

from app.models.users import User


def insert_user(name: str) -> User:

    user = User(name)

    with session_scope() as session:

        session.add(user)

        session.flush()

        return user


def fetch_user(id: Optional[int]=None, name: Optional[str]=None):

    with session_scope() as session:

        query = session.query(User)

        if id:

            query = query.filter(User.id == id)

        if name:

            query = query.filter(User.name == name)

        return query.all()

As the name suggests, insert_user performs an INSERT and fetch_user performs a SELECT.


schema

Finally, we come to the GraphQL part.


users.py

In users.py and todos.py, we generate classes for GraphQL corresponding to the classes generated by models earlier.


users.py

import graphene

from graphene import relay

from graphene_sqlalchemy import SQLAlchemyObjectType


from app.models.users import User as UserModel

from app.cruds.users import insert_user



#-------------------------

# Query

#-------------------------

class User(SQLAlchemyObjectType):

    class Meta:

        model = UserModel

        interface = (relay.Node, )


class UserConnections(relay.Connection):

    class Meta:

        node = User


#-------------------------

# Mutation.

#-------------------------

class InsertUser(relay.ClientIDMutation):

    class Input:

        String(required=True)


    User = graphene.Field(User)


    @classmethod

    def mutate_and_get_payload(cls, root, info, name):

        user = insert_user(name)

        return InsertUser(user)

Query

class User is generated by inheriting from SQLAlchemyObjectType.

In addition, it provides a good connection between GraphQL and SQLAlchemy by tying the SQLAlchemy class with model = UserModel.

See the graphene-sqlalchemy repository for details.


In addition, you can easily provide Relay support by defining interface = (relay.Node, ) and UserConnections.

For more information, please refer to the graphene relay documentation.


Mutation

class InsertUser is generated by inheriting from relay.ClientIDMutation.

Define the arguments for executing InsertUser with Input.

In Mutate_and_get_payload, define the process of actually accessing the DB. (We will use the insert_user function defined in users in cruds earlier.)


schema.py

In schema.py, we define the root part of Query and Mutation.


schema.py

import graphene

from graphene import relay

from graphene_sqlalchemy import SQLAlchemyObjectType, SQLAlchemyConnectionField


from app.cruds.users import fetch_user

from app.cruds.todos import fetch_todo

from app.schema.users import User, UserConnections, InsertUser

from app.schema.todos import Todo, TodoConnections, InsertTodo


#-------------------------

# Query

#-------------------------

class Query(graphene.ObjectType):

    Node = relay.Node.Field()


    user = graphene.Field(

        lambda: graphene.List(User),

        id=graphene.Int(required=False),

        String(required=False), name=graphene.String(required=False)

    )

    todo = graphene.Field(

        lambda: graphene.List(Todo),

        List(Todo), id=graphene.Int(required=False),

        List(Todo), id=graphene.Int(required=False), user_id=graphene.Int(required=False),

        String(required=False), user_id=graphene.Int(required=False), title=graphene.String(required=False),

        String(required=False), title=graphene.String(required=False), description=graphene,

        DateTime(required=False), deadline=graphene.DateTime(required=False)

    )

    all_users = SQLAlchemyConnectionField(UserConnections)

    all_todos = SQLAlchemyConnectionField(TodoConnections, sort=None)


    def resolve_user(self, info, id=None, name=None):

        return fetch_user(id, name)


    def resolve_todo(self, info, id=None, user_id=None, title=None, description=None, deadline=None):

        return fetch_todo(id, user_id, title, description, deadline)


#-------------------------

# Mutation

#-------------------------

class Mutation(graphene.ObjectType):

    Insert_user = InsertUser.Field()

    Field(): insert_todo = InsertTodo.


#-------------------------

# Schema

#-------------------------

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

Query

node = relay.Node.Field()

all_users = SQLAlchemyConnectionField(UserConnections)

all_todos = SQLAlchemyConnectionField(TodoConnections, sort=None)

is to be able to throw queries according to the RELAY standard.


user and resolve_user and todo and resolve_todo are the counterparts, which allow the SELECT process defined in the manual to be handled as GraphQL.


Mutation

Mutation classes generated by users.py and todos.py are connected to the Mutation class.


Schema

We will bind Root's Query class and Mutation class to Schema.


main.py

Finally, we generate FastAPI endpoints.


main.py

import sys, pathlib, uvicorn, graphene


current_dir = pathlib.Path(__file__).resolve().parent

sys.path.append( str(current_dir) + '/. /' )


from graphql.execution.executors.asyncio import AsyncioExecutor

from fastapi import FastAPI

from starlette.graphql import GraphQLApp

from app.schema.schema import schema


app = FastAPI()

app.add_route("/", GraphQLApp(schema=schema))


if __name__ == "__main__":

    uvicorn.run("main:app", port=5000, reload=True, access_log=False)

Create an app instance with FastAPI() and configure route settings to access GraphQL with /.


Execution

You can start it with the following command



python . /app/main.py

Confirmation of results

This time, we will check the results using GraphQL Playground.


Schema

The results are displayed as follows on GraphQL Playground.

Screenshot 2020-12-14 22.36.03.png


The whole Schema looks like this


scalar DateTime


input InsertTodoInput {

  userId: Int!

  title: String!

  description: String!

  deadline: DateTime!

  clientMutationId: String!

}


type InsertTodoPayload {

  todo: Todo

  clientMutationId: String

}


input InsertUserInput {

  name: String!

  clientMutationId: String

}


type InsertUserPayload {

  user: User

  clientMutationId: String

}


type Mutation {

  insertUser(input: InsertUserInput!): InsertUserPayload

  insertTodo(input: InsertTodoInput!): InsertTodoPayload

}


# An object with an ID

interface Node {

  # An object with an ID interface Node { # The ID of the object.

  id: ID!

}


# The Relay compliant `PageInfo` type, containing data necessary to paginate this connection.

type PageInfo {

  # When paginating forwards, are there more items?

  hasNextPage: Boolean!

  # When paginating backwards, are there more items?

  hasPreviousPage: Boolean!

  # When paginating backwards, the cursor to continue.

  startCursor: String

  # When paginating forwards, the cursor to continue. startCursor: String!

  endCursor: String

}


{ type Query {

  # The ID of the object

  node(id: ID!): Node

  user(id: Int, name: String): [User].

  # <graphene.types.scalars.String object at 0x111cd6ac0>

  todo(id: Int, userId: Int, title: String, deadline: DateTime): [Todo].

  allUsers(

    sort: [UserSortEnum].

    before: String

    after: String

    first: Int

    last: Int

  ): UserConnectionsConnection

  allTodos(

    before: String

    after: String

    first: Int

    last: Int

  ): TodoConnectionsConnection

}


type Todo {

  id: ID!

  userId: Int

  title: String

  description: String

  finished: Boolean

  deadline: DateTime

  users: User

}


type TodoConnectionsConnection {

  # Pagination data for this connection.

  pageInfo: PageInfo!

  # Contains the nodes in this connection.

  edges: [TodoConnectionsEdge]!

}


# A Relay edge containing a `TodoConnections` and its cursor.

type TodoConnectionsEdge {

  # The item at the end of the edge

  node: Todo

  # A cursor for use in pagination

  cursor: String!

}


} type User {

  id: ID!

  name: String

}


type UserConnectionsConnection {

  # Pagination data for this connection.

  pageInfo: PageInfo!

  # Contains the nodes in this connection.

  edges: [UserConnectionsEdge]!

}


# A Relay edge containing a `UserConnections` and its cursor.

type UserConnectionsEdge {

  # The item at the end of the edge

  node: User

  # A cursor for use in pagination

  cursor: String!

}


# An enumeration.

enum UserSortEnum {

  ID_ASC

  ID_DESC

  NAME_ASC

  NAME_DESC

}

PageInfo and interface Node are related to Relay.


We will also look at the other results of the query.


allUsers

Screenshot 2020-12-14 22.42.41.png


allTodos

Screenshot 2020-12-14 22.44.18.png


user(resolve_user)

Screenshot 2020-12-14 22.45.08.png


todo(resolve_todo)

Screenshot 2020-12-14 22.46.20.png


insertUser

Screenshot 2020-12-14 22.49.53.png


insertTodo

Screenshot 2020-12-14 22.53.19.png


Conclusion

This is my first time to build the server side of GrpahQL, but I feel it is more difficult than REST API because I am not used to it.

I will ask my fellow engineers to show me how to do it.


It has been a while since I wrote the code, so I may not have written it very neatly, or I may have misunderstood some parts, in which case I would appreciate it if you could tell me in the comments section.


Tomorrow, @ham0215 will write an article about the GraphQL conversion of ideagram from scratch, so stay tuned!


Reference

Official

SQLAlchemy 1.3 Documentation

Graphene Documentaion

FastAPI Documentaion

graphene-sqlalchemy repository


Articles

How to handle session in SQLAlchemy

GraphQL Tutorial in Python + α

FastAPI Directory Design

sqlalchemy table definition tips

0 コメント:

コメントを投稿