GraphQL and MongoDB in Python 🐍

GraphQL is a Query language for dealing with your API. The best part is that it can fetch multiple API at ones with a single hit!

In Python, graphene library is used to work with GraphQL in Python for more details visit(https://graphene-python.org). so in this coding blog, we will see how we can connect with the MongoDB database and get the result of multiple queries with a single hit to sever.

Before writing any code we need to install python server for that I will use aiohttp framework. for more detail on aiohttp visit(https://aiohttp.readthedocs.io/en/stable/)

pip install aiohttp

Its time to check aiohttp is running properly!

create a new python file with .py ..i.e filename.py

from aiohttp import web
import json
async def apitest(request):
    
    result ={"status":"200"}
    return web.Response(text=json.dumps(result), status=200)

app = web.Application()
app.router.add_get('/apitest',apitest)
web.run_app(app)

Run file python filename.py in terminal

As the server is setup properly now its time to create user collection in MongoDB.

Pip the following dependencies.

• pip install graphene_mongo
• pip install graphene

Models are required in GraphQL for binding data so create model.py

from mongoengine import Document
from mongoengine.fields import(StringField,ListField,ReferenceField)
class User(Document):
    meta = {'collection': 'user'}
    firstName = StringField(required=True)
    lastName = StringField(required=True)
    type = StringField(required=True)

After creating models its time to create a schema name this file as api.py

import json
import graphene
import model as UserModel
from mongoengine import connect
from graphene_mongo import MongoengineObjectType
from aiohttp import web

async def mongo(request):
    data = request
    connect('database', host='localhost' ,port=27017)
    class User(MongoengineObjectType):
        class Meta:
            model = UserModel.User
    class Query(graphene.ObjectType):
       users = graphene.List(User)
       usertype = graphene.List(User,types=graphene.String())

       def resolve_users(self, info):
         return list(UserModel.User.objects.all())

       def resolve_usertype(self, info,types):
         return list(UserModel.User.objects.filter(type=types))


    schema = graphene.Schema(query=Query,types=[User])
    query = request.query['query']
    result = schema.execute(query)
    return web.Response(text=json.dumps(result.data), status=200)


app = web.Application()
app.router.add_get('/mongo', mongo)
web.run_app(app)

Fetch all users

query { users { firstName lastName type } }

Fetch user by there type

query { usertype( types : "admin") { firstName lastName type } } 
or 
query { usertype( types : "editor"){ firstName lastName type } }

Fetch all users and type admin at sametime

query { 
        usertype( types : "admin") { firstName lastName type }
        users{ firstName lastName } 
      }

Feel free to share and comment down-below!