Creating Dynamic Schema on Runtime Graphene

Question

I almost spent 3 days to find a way for creating a dynamic schema in python graphene. the only related result I could find is the below link: https://github.com/graphql-python/graphene/blob/master/graphene/types/dynamic.py But I couldn't find any documentation for it.

The whole idea is to create a dynamic schema. I want to provide a GraphQL compatible API that makes users able to query my contents even if Models are not defined in the code. In other words, I want to create Models on the fly. I have no idea about what shall I do.

It would be a great favor if you can provide an example for that.

Update :

My Project is a Headless CMS which has a feature that users can create their own content types and I want to provide a GraphQL interface to make everything easier and more flexible.

Here is example of my Content Types in DB :

{
  "id": "author",
  "name": "Book Author",
  "desc": "",
  "options":[
    {
      "id": "author_faname",
      "label": "Sample Sample",
      "type": "text",
      "required": true,
      "placeholder":"One Two Three Four"
    },
    {
      "id": "author_enname",
      "label": "Sample label",
      "type": "text",
      "required": true,
      "placeholder":"Sample Placeholder"
    }
  ]
}

And Here is Stored content in DB based on that content type :

{
  "id": "9rqgbrox10",
  "content_type": "author",
  "data":{
    "author_fname":"Jimmy",
    "author_ename":"Hello"
  }
}

Now as my Models are not declared in Code and they are completely in DB, I want to make my schemas on the fly and I don't know what is best the solution for this. I know there should be a way because the other Headless CMS Projects are providing this.

Thanks in advance!

Answer 1

Basically, schema is created something like this:

class MyType(graphene.ObjectType):
    something = graphene.String()

class Query(graphene.ObjectType):
    value = graphene.Field(MyType)

schema = graphene.Schema(query=Query, types=[MyType])

First, in order to add some kind of dynamics, you will most likely want to wrap the above code in a function like create_schema().

Then, when you want to dynamically create a class during runtime, the above code can be rewritten like this:

def create_schema():
    MyType = type('MyType', (graphene.ObjectType,), {
        'something': graphene.String(),
    })

    Query = type('Query', (graphene.ObjectType,), {
        'value': graphene.Field(MyType),
    })

    return graphene.Schema(query=Query, types=[MyType])

For your example it could look something like this:

def make_resolver(record_name, record_cls):
    def resolver(self, info):
        data = ...
        return record_cls(...)
    resolver.__name__ = 'resolve_%s' % record_name
    return resolver

def create_schema(db):
    record_schemas = {}
    for record_type in db.get_record_types():
        classname = record_type['id'].title()  # 'Author'
        fields = {}
        for option in record_type['options']:
            field_type = {
                'text': graphene.String,
                ...
            }[option['type']
            fields[option['id']] = field_type()  # maybe add label as description?
        rec_cls = type(
            classname,
            (graphene.ObjectType,), 
            fields,
            name=record_type['name'],
            description=record_type['desc'],
        )
        record_schemas[record_type['id']] = rec_cls

    # create Query in similar way
    fields = {}
    for key, rec in record_schemas:
        fields[key] = graphene.Field(rec)
        fields['resolve_%s' % key] = make_resolver(key, rec)
    Query = type('Query', (graphene.ObjectType,), fields)

    return graphene.Schema(query=Query, types=list(record_schemas.values()))

Note that if you try to insert new fields into already existing class, like this - MyType.another_field = graphene.String(), then it won't work: that is because when graphene.ObjectType class is instantiated, all its fields are recorded in self._meta.fields OrderedDict. And updating it is not as straightforward as just MyType._meta.fields['another_field'] = thefield - see the code of graphene.ObjectType.__init_subclass_with_meta__ for details.

So if your schema is dynamically changed then it might be better to fully re-create it from scratch than to patch it.

Answer 2

I'd like to share another neat method.

So, the issue is that the graphene.ObjectType is not a regular Python class. It has a special metaclass that you can see implemented here. At the moment Python takes care of the inheritance process (when the class itself is initalized), graphene does some operations to register the type. I didn't find a way to change types after the inheritance happens. However, if you just want to generate the schema out of a predefined boilerplate (like me) or some other source, you can do something like this. I first define a handy inherit method:

def inherit_from(Child, Parent, persist_meta=False):
    """Return a class that is equivalent to Child(Parent) including Parent bases."""
    PersistMeta = copy(Child.Meta) if hasattr(Child, 'Meta') else None

    if persist_meta:
        Child.Meta = PersistMeta

    # Prepare bases
    child_bases = inspect.getmro(Child)
    parent_bases = inspect.getmro(Parent)
    bases = tuple([item for item in parent_bases if item not in child_bases]) + child_bases

    # Construct the new return type
    try:
        Child = type(Child.__name__, bases, Child.__dict__.copy())
    except AttributeError as e:
        if str(e) == 'Meta':
            raise AttributeError('Attribute Error in graphene library. Try setting persist_meta=True in the inherit_from method call.')
        raise e

    if persist_meta:
        Child.Meta = PersistMeta

    return Child

Now the key is to perform the inheritance when the type's class is not about to change anymore.

def context_resolver_factory(attr):
    """Create a simple resolver method with default return value None."""

    def resolver(obj, info):
        return info.context.get(attr, None)

    return resolver


class User:
    id = graphene.ID()
    name = graphene.String(resolver=lambda user, info: user.name)


class Query: pass
    me = graphene.Field(User)

    def resolve_me(self, info):
        return info.context["user"]


inherit_from(User, graphene.ObjectType)  # no changes to User class are possible after this line

# method 1: sometimes it's really neat and clean to include a resolver in the field definition
setattr(Query, 'user', graphene.User(resolver=context_resolver_factory('user'))
# or even use lambda if a factory is still overkill
setattr(Query, 'user', graphene.User(resolver=lambda query, info: info.context["user"]))


# method 2: if you want to set the resolver separately, you can do it this way
setattr(Query, 'user', graphene.User())
setattr(Query, 'resolve_user', context_resolver_factory('user'))

# any changes to `Query.Meta` can be done here too

inherit_from(Query, graphene.ObjectType)  # no changes to Query class are possible after this line

schema = graphene.Schema(query=Query)

Where I got with my little library was generating everything from a boilerplate class like this:

@register_type('Product')
class ProductType:
    class Meta:
        model = Product
        fields = '__all__'
        related_fields = {
            NestedField('tags', TagType),
            NestedField('related_products', 'self'),
        }
        lookups = {
            'id': graphene.ID(),
            'name': graphene.String(description="Name"),
            'ean': graphene.String(),
            'brand': graphene.String()
        }
        filters = {
            'ids': IDFilter,
            'django_filter': DjangoFilter,
            'pagination': PaginationFilter,
            'search_name': ProductMLNSearchFilter
        }

Biggest challenge were the NestedFields and figuring out automatic Django ORM query select/prefetch when a request comes in, but I won't go into detail unless that's something relevant.