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!
0 Comments