In GraphQL, we can do multiple inserts, update and delete at a time. The cool thing about GraphQL is we have a lot of control in our hands. In the GraphQL query, we can ask exactly what we need. As well as query, we can post as much data as we want with GraphQL mutation.

There are a lot of real-life scenarios where we need a bulk insert, update or delete. Specially, In our modern Single Page Applications (SPA), a user may create, update or delete multiple related items at a time. Though GraphQL makes it easy to send data for bulk upsert or delete, we need a fast, cleaner way to do it with Django. The main challenge here is, we have to do a faster, memory-efficient way to implement upsert with large batch size.

Basic Setup

In our previous post, we created a simple blog app with Django-Graphene. To demonstrate upsert with GraphQL, we will add another model called BlogTag. To add tags for Blog add the following code to file.

class BlogTag(models.Model):
    blog = models.ForeignKey(Blog, on_delete=models.CASCADE, related_name="blog_tags")
    tag = models.CharField(max_length=250) 

Now run the following commands to generate migrations.

python makemigrations

python migrate

And create a file called blog_tags in the blog/schema directory.

Here, add the following schema for blog tags.

import graphene
from graphene_django.types import DjangoObjectType, ObjectType
from ..models import BlogTag

class BlogTagFields():
    blog_id = graphene.ID()
    tag = graphene.String()
    id = graphene.ID()

class BlogTagType(DjangoObjectType, BlogTagFields):
    class Meta:
        model = BlogTag

class BlogTagInputType(graphene.InputObjectType, BlogTagFields):

In schema, import BlogTagType and BlogTagInputType as following.

from .blog_tags import BlogTagType, BlogTagInputType

And update BlogInputType as following.

class BlogInputType(graphene.InputObjectType, BlogFields):
    id = graphene.ID()
    author_id = graphene.ID()
    tags = BlogTagInputType()
    deleted_tags = BlogTagInputType()

Also, update BlogType as following.

class BlogType(DjangoObjectType, BlogFields):
    class Meta:
        model = Blog

    id = graphene.ID(required=True)
    author = AuthorType()
    blog_tags = BlogTagType()

Upsert/Delete Manager

We will create a utility manager for upsert/delete functionality which will work as a base class. To do that open blog/ and add the following manager.

from itertools import islice

class BulkManager(object):
    def __init__(self, model, items=[], update_dict={}, removed_items=[], batch_size=100):

        self.model = model
        self.update_dict = update_dict
        self.batch_size = batch_size
        self.removed_items = removed_items

        self.new_items = list(filter(lambda item: not item.get("id"), items))
        self.existing_items = list(
            filter(lambda item: item.get("id") is not None, items)

        if self.new_items:
        if self.existing_items:
        if self.removed_items:

    def _prepare_bulk_create_data(self):
        for item in self.new_items:
            yield self.model(**item)

    def _prepare_bulk_update_data(self):
        for item in self.existing_items:
            yield self.model(**item)

    def bulk_create(self):
        data = self._prepare_bulk_create_data()
        while True:
            items = list(islice(data, self.batch_size))
            if not items:
            self.model.objects.bulk_create(items, self.batch_size)

    def bulk_update(self):
        field_item = self.existing_items[0]
        fields = list(filter(lambda key: key is not "id", field_item.keys()))
        data = self._prepare_bulk_update_data()

        while True:
            items = list(islice(data, self.batch_size))
            if not items:
            self.model.objects.bulk_update(items, fields)

    def bulk_delete(self):
        ids = [item.get("id") for item in self.removed_items if item.get("id")]
        del_objs = self.model.objects.filter(pk__in=ids)
        if del_objs.exists():

Now go to blog/ file and import BulkManager as following.

from .utils import FormValidator, BulkManager

And add the following code at the end of the save method of the CreateBlogForm.


Your CreateBlogForm should look like this.

from .utils import FormValidator, BulkManager
from .models import Blog, BlogTag, Author

class CreateBlogForm(FormValidator):
    def validate_author_id(self):
        author_id ="author_id", None)
        if not author_id:
            return self.errors.update({"author_id":"The author_id feild is required."})

        if not Author.objects.filter(id=author_id).exists():
            return self.errors.update({"author_id":"Author {author_id} does not exist".format(author_id=author_id)})
        self.cleaned_data["author_id"] = author_id         

    def validate_title(self):
        title ="title", None)
        if not title:
            return self.errors.update({"title":"The title feild is required."})
        if len(title) < 10:
            return self.errors.update({"title":"Title is too short."})
        self.cleaned_data["title"] = title 

    def save(self):
        data =

        blog = Blog()
        for key, val in data.items():
            setattr(blog, key, val)

        return blog

All right. We are done. It’s time to test the bulk manager.

Test bulk upsert/delete manager

To test the bulk manager, copy this URL to your browser.

Add the following mutation to GraphiQL console and run it.

mutation createBlog {
  createBlog(input: {
    title: "How to build GraphQL API with Django.",
    authorId: 1,
    body: "GraphQL is a declarative, strongly typed, data-driven query language to build APIs.",
    tags: [
      {tag: "GraphQL"},
      {tag: "Graphene-Django"},
      {tag: "Bulk Upsert"},
      {tag: "Django Upsert"},
      {tag: "GraphQL Upsert"},
      {tag: "Django Upsert/Delete"}]
  }) {
    blog {
      author {
      blogTags {

You should see the following result.

  "data": {
    "createBlog": {
      "ok": true,
      "blog": {
        "id": "8",
        "title": "How to build GraphQL API with Django.",
        "author": {
          "id": "1",
          "name": "Guido van Rossum"
        "body": "GraphQL is a declarative, strongly typed, data-driven query language to build APIs.",
        "blogTags": [
            "id": "19",
            "tag": "GraphQL"
            "id": "20",
            "tag": "Graphene-Django"
            "id": "21",
            "tag": "Bulk Upsert"
            "id": "22",
            "tag": "Django Upsert"
            "id": "23",
            "tag": "GraphQL Upsert"
            "id": "24",
            "tag": "Django Upsert/Delete"


It’s very common that if you work with Django for a while, you may need to do upsert/delete for your project. We need to implement this with a clean, fast and efficient way to get better performance for large batches.

Here we added a BulkManager to do upsert/delete with Graphene-Django. If you would like to see the full source code, please check it out from my repo.

+ posts

Full-stack Developer (Python | Django | React | React-Native | Angular | Vue)