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 コメント:
コメントを投稿