[FastAPI] 3. SQLAlchemy + Graphene 조합으로 GraphQL 서버 만들기

이번 포스트에서는 GraphQL에 대한 이야기를 해보도록 하겠습니다.

 

 

 

 

What is GraphQL ?

GraphQL에 대한 이야기는 그리 길지 않기 때문에 여기서 간단하게 다뤄보도록 하겠습니다.

 

GraphQL은 Facebook에서 개발한 데이터 표현 언어로 여기서 QL이 Query Language에 해당합니다. 우리는 서버에서 데이터를 가져오기 위해 REST API를 사용하였고, REST API는 메소드와 요청 데이터 혹은 응답 데이터로 구분하여 원하는 데이터를 가져오거나 데이터를 적재하는 등을 수행하였습니다.

 

그러나 GraphQL은 메소드 없이 오직 Query Language만을 이용하여 원하는 모델을 적재하고, 가져올 수 있습니다. 또한 원하는 항목을 클라이언트가 직접 명시함으로써 서버에서 제공해주는 데이터만을 의존하지 않고 원하는 데이터를 가져올 수 있는 장점이 생기며 이 장점은 차후 네트워크 대역폭을 절약시키는 데 도움을 주기도 합니다.

 

이 외에도 다양한 장점이 존재합니다.

 

• UnderFetching
  
  HTTP 요청 1번에 원하는 데이터를 모두 가져올 수 있음.
  
  
• OverFetching
  
  클라이언트가 원하는 항목만 명시하면 해당 데이터만을 반환 (대역폭 절약, 그러나 약간의 오버헤드 발생)
  
  
• Type Specification
  
  요청과 응답에 대해 자료형을 지정하는 것을 필수로 하며 이로 인한 처리를 유연하게 할 수 있음.
  
  
• Cross Platform
  
  Web, Mobile을 구분하지 않고, 하나의 API로 모든 작업을 처리함.

타입을 명시하는 것에 대해서는 조금 의아할 수 있겠지만 Python에서는 기본적으로 DTO를 갖추지 않아도 됩니다. 그러나 DTO를 갖추지 않으면 각 API별로 어떻게 요청을 해야할지, 응답을 주는지가 명시되어 있지 않아 사용하는 사람 입장에서는 오히려 혼란을 야기할 수도 있는 부분입니다. 

 

그러나 GraphQL에서는 이러한 타입 정의가 기본이며 파라미터 값의 자료형이 다를 경우 422 오류를 반환합니다. 

 

 

 

 

Graphene

Graphene는 Python에서 GraphQL을 사용하기 위한 라이브러리입니다. 아마 Node.js에서 GraphQL을 써보신 분들은 Apollo 라는 것을 들어보셨을 것입니다. Python에서 이와 비슷한 Ariadne는 스키마 우선 접근 방식을 사용함으로써 Python 코드가 아닌 SDL(Schema Definition Language)에 의존하여 사용하기 때문에 일반적으로는 접근하기 어려운 방식입니다.

 

그러나 Graphene는 프로그래밍 코드 우선 접근 방식을 사용함으로써 GraphQL을 처음 사용해보시는 분들에게 유리합니다. Schema에 대한 개념이나 코드를 잘 몰라도 Python 코드만 짤 줄 안다면 GraphQL을 사용할 수 있고, 거기에 SQLAlchemy라는 ORM과도 쉽게 통합할 수 있어, 매우 빠른 개발에 용이한 라이브러리입니다.

 

Graphene-Python

Graphene는 보다시피 단독으로 사용할 수도 있고, Flask와 통합하거나, SQLAlchemy와 통합하는 등 다양한 방법으로 사용할 수 있습니다.

 

이 중에서도 이 포스트에서는 SQLAlchemy와 통합하여 사용하는 방법에 대해 알아보겠습니다.

 

 

 

 

SQLAlchemy + Graphene

지난 포스트에서 사용했던 프로젝트와 코드를 그대로 사용하여 아래처럼 graphene-sqlalchemy 디펜던시만 추가해주도록 하겠습니다.

[tool.poetry]
name = "fastapiexample"
version = "0.1.0"
description = ""
authors = ["Neon K.I.D <contact@neonkid.xyz>"]

[tool.poetry.dependencies]
python = "^3.8.5"
fastapi = "^0.63.0"
uvicorn = "^0.13.2"
SQLAlchemy = "^1.3.22"
psycopg2-binary = "^2.8.6"
graphene-sqlalchemy = "^2.3.0"

[tool.poetry.dev-dependencies]

[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"

기존의 코드를 이어서 사용하기 때문에 REST API 코드와 겹칠 것입니다. 하지만 상관없습니다. 이 포스트를 그대로 따라한다면 GraphQL과 REST API를 동시에 사용할 수도 있습니다.

 

우리는 지난 시간에 간단한 메모를 할 수 있는 DB 스키마를 만들었고, 여기에 메모의 제목과 컨텐츠, 즐겨찾기 여부 등을 확인할 수 있도록 하였습니다.

 

이를 토대로 GraphQL 기반의 API를 만들어보도록 하겠습니다.

 

 

 

Resolver

ObjectType은 graphene에서 GraphQL을 구현하기 위한 기본적인 Query 클래스로 GraphQL API를 만들기 위해 반드시 만들어야 할 필수 요소입니다.

from graphene import ObjectType


class Query(ObjectType):
	

relay를 사용하여 DB 서버와 연결할 수 있습니다.

from graphene import ObjectType, relay


class Query(ObjectType):
    node = relay.Node.Field()

먼저 Query에 node라는 변수를 담고, relay 필드임을 명시합니다. 이렇게 하면 앞으로 GraphQL에서 node 밑으로 DB에서 불러온 데이터가 들어오게 됩니다.

 

그리고 이 DB와 연결할 수 있는 커넥션을 하나 만들어줍니다.

from graphene import relay
from graphene_sqlalchemy import SQLAlchemyObjectType


class MemoModel(SQLAlchemyObjectType):
    class Meta:
        model = Memo
        interfaces = (relay.Node,)


class MemoConnection(relay.Connection):
    class Meta:
        node = MemoModel

relay에 있는 Connection을 상속받고, 이에 해당하는 모델을 node에 넣어줍니다.

from graphene import ObjectType, relay, String
from graphene_sqlalchemy import SQLAlchemyObjectType, SQLAlchemyConnectionField


class Query(ObjectType):
    node = relay.Node.Field()

    memo = SQLAlchemyConnectionField(MemoConnection, id=String())
    memo_list = SQLAlchemyConnectionField(MemoConnection, sort=MemoModel.sort_argument())

    def resolve_memo(self, info, **kwargs):
        id = kwargs.get('id')

        memos_query = MemoModel.get_query(info)

        if id is not None:
            return memos_query.filter_by(id=id)

    def resolve_memo_list(self, info, **kwargs):
        return MemoModel.get_query(info).all()

마지막으로 아까 위에서 만든 Query에 memo와 memo_list에 대한 내용을 넣어줍니다. memo는 단일 메모를 가져오기 위한 파라미터이고, memo_list는 복수 개의 메모를 가져오기 위한 파라미터입니다. 우린 ID로 조회를 하는 REST API를 만들었기 때문에 id 검색을 위한 id 파라미터 값을 만들어줍니다.

 

그리고 전체 메모 조회시 정렬 알고리즘을 사용하기 위해 memo_list에는 sort 아규먼트를 하나 추가해줍니다.

from graphene import Schema
from starlette.graphql import GraphQLApp


app.add_route('/graphql', GraphQLApp(schema=Schema(query=Query)))

이제 /graphql 이라는 엔드포인트에 GraphQL API를 묶어주면 graphql 엔드포인트 하나로 데이터를 READ 할 수 있습니다.

 

 

 

 

Mutation

Mutation은 GraphQL에서 Collection을 새로 만들고, 수정할 때 사용하는 ObjectType입니다. graphene에서도 GraphQL로 데이터를 INSERT, UPDATE할 때 이를 사용합니다.

from graphene import String, Mutation, Boolean


class InsertMemo(Mutation):
    class Arguments:
        title = String(required=True)
        content = String()
        is_favorite = Boolean()
        
    memo = Field(lambda: MemoModel)
    
    def mutate(self, info, **kwargs):
        memo = Memo(**kwargs)
        
        db_session.add(memo)
        db_session.commit()
        
        return InsertMemo(memo)

먼저 기본적으로 메모를 삽입할 때는 Mutation을 상속 받고, 그에 사용할 Argument를 정의해줍니다. 타입은 graphene 라이브러리에서 제공되는 것으로 사용합니다.

 

그리고 SQLAlchemy의 세션을 이용하여 DB에 영속하도록 합니다.

from graphene import String, Mutation, Boolean, Field


class UpdateMemo(Mutation):
    class Arguments:
        id = String(required=True)
        title = String()
        content = String()
        is_favorite = Boolean()

    memo = Field(lambda: MemoModel)

    def mutate(self, info, **kwargs):
        memo = db_session.query(Memo).filter_by(id=kwargs.get('id')).first()

        for key, value in kwargs.items():
            setattr(memo, key, value)

        db_session.commit()
        
        return UpdateMemo(memo)

비슷한 방법으로 메모를 업데이트 할 때 위와 같이 만들어줍니다.

from graphene import String, Mutation, Boolean


class DeleteMemo(Mutation):
    class Arguments:
        id = String(required=True)
    
    def mutate(self, info, id):
        memo = db_session.query(Memo).filter_by(id=id)
        db_session.delete(memo)
        return ""

삭제를 할 때도 동일하게 개발하면 됩니다.

from graphene import ObjectType


class Mutation(ObjectType):
    create_memo = InsertMemo.Field()
    update_memo = UpdateMemo.Field()
    delete_memo = DeleteMemo.Field()

GraphQL App 클래스에 담을 Mutation ObjectType을 하나 만들어주고, 아규먼트에 각각 만들어준 API를 넣습니다.

from graphene import Schema
from starlette.graphql import GraphQLApp


app.add_route('/graphql', GraphQLApp(schema=Schema(query=Query, mutation=Mutation, types=[MemoModel])))

마지막으로 GraphQLApp 클래스에 Mutation과 사용할 Type을 넣습니다. 사용할 Type을 넣을 때는 grpahene 라이브러리에 있는 클래스로 상속 받은 것을 넣어야 합니다.

 

 

 

 

 

Test

테스트에 대해서는 간단하게 Resolver와 Mutation만을 가지고 진행하도록 하겠습니다. 이 역시 Postman을 사용합니다.

mutation {
	createMemo(title: "", content: "") {
		memo {
			id
			title
			content
		}
	}
}

Mutation에 대한 테스트입니다. GraphQL을 사용할 때 mutation임을 선언하고, 코드에서 정의한 함수를 호출합니다. 그리고 반환할 데이터 타입을 부른 후 반환하고자 하는 타입을 명시하면 됩니다.

query {
	memoList {
		edges {
			node {
				id
				title
				content
			}
		}
	}
}

반대로 DB에 적재된 데이터를 호출하고자 할 때는 query를 선언해줍니다. 그리고 메모 리스트를 불러오는 memoList를 정의하고 그에 해당하는 하위 데이터를 부르기 위해 edges와 node를 입력한 다음 원하는 아규먼트를 내리면 됩니다.

 

 

 

 

마치며...

여기까지 FastAPI + SQLAlchemy + Grpahene 조합으로 간단한 GraphQL API를 만드는 법까지 알아봤습니다. GraphQL을 아주 간단하게 장단과 차이점만 대해서 설명하고 넘어갔기 때문에 초면에 바로 사용하는 것은 어려울 수도 있을 것입니다.

 

그러나 보다시피 REST API에 비해 API 구조가 간단하고, 클라이언트가 원하는 데이터를 뽑을 수 있다는 것은 대역폭을 줄이고, 호출 횟수가 낮아진다는 점에 있어 매우 큰 이득을 볼 수 있습니다.

 

FastAPI에서 기존에 사용하던 REST API 그대로 GraphQLApp 인스턴스만을 만들어 GraphQL API를 만들 수 있기 때문에 기존에 REST API를 만들어 사용하다가 GraphQL로 전환을 쉽게 할 수 있습니다.