Skip to content

strollby/graphene-mongo

 
 

Repository files navigation

Build Status Coverage Status Documentation Status PyPI version PyPI pyversions Downloads

Lint Test Package

Graphene-Mongo

A Mongoengine integration for Graphene.

Installation

For installing graphene-mongo, just run this command in your shell

pip install graphene-mongo

Examples

Here is a simple Mongoengine model as models.py:

from mongoengine import Document
from mongoengine.fields import StringField


class User(Document):
    meta = {'collection': 'user'}
    first_name = StringField(required=True)
    last_name = StringField(required=True)

To create a GraphQL schema and sync executor; for it you simply have to write the following:

import graphene

from graphene_mongo import MongoengineObjectType

from .models import User as UserModel


class User(MongoengineObjectType):
    class Meta:
        model = UserModel


class Query(graphene.ObjectType):
    users = graphene.List(User)

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


schema = graphene.Schema(query=Query)

Then you can simply query the schema:

query = '''
    query {
        users {
            firstName,
            lastName
        }
    }
'''
result = await schema.execute(query)

To create a GraphQL schema and async executor; for it you simply have to write the following:

import graphene

from graphene_mongo import AsyncMongoengineObjectType
from graphene_mongo.utils import sync_to_async
from concurrent.futures import ThreadPoolExecutor

from .models import User as UserModel


class User(AsyncMongoengineObjectType):
    class Meta:
        model = UserModel


class Query(graphene.ObjectType):
    users = graphene.List(User)

    async def resolve_users(self, info):
        return await sync_to_async(list, thread_sensitive=False,
                             executor=ThreadPoolExecutor())(UserModel.objects.all())


schema = graphene.Schema(query=Query)

Then you can simply query the schema:

query = '''
    query {
        users {
            firstName,
            lastName
        }
    }
'''
result = await schema.execute_async(query)

To learn more check out the following examples:

Contributing

After cloning this repo, ensure dependencies are installed by running:

pip install -r requirements.txt

After developing, the full test suite can be evaluated by running:

make test