Skip to content

Instantly share code, notes, and snippets.

@dorosch
Created April 13, 2020 07:08
Show Gist options
  • Star 3 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save dorosch/6cffd2936ac05ef8794c82901ab4d6e7 to your computer and use it in GitHub Desktop.
Save dorosch/6cffd2936ac05ef8794c82901ab4d6e7 to your computer and use it in GitHub Desktop.
Django bulk get_or_create manager
from django.db import models
class BulkGetOrCreateManager(models.Manager):
def bulk_get_or_create(self, objs, lookup_field=None):
assert lookup_field, "Not set 'lookup_field' for 'bulk_get_or_create'"
lookup = {f'{lookup_field}__in': objs}
existing_objects = [
obj for obj in self.get_queryset().filter(**lookup)
]
non_existing_objects = [
obj for obj in objs if obj not in existing_objects
]
self.bulk_create(non_existing_objects, batch_size=999)
return super().get_queryset().filter(**lookup)
@sevazhidkov
Copy link

sevazhidkov commented Nov 26, 2020

This manager seemed to not be working properly. Here is mine slightly modified version:

class BulkGetOrCreateManager(models.Manager):
    def bulk_get_or_create(self, objs, lookup_field=None):
        assert lookup_field, "Not set 'lookup_field' for 'bulk_get_or_create'"

        lookup = {f'{lookup_field}__in': [getattr(obj, lookup_field) for obj in objs]}
        existing_objects = [
            obj for obj in self.get_queryset().filter(**lookup)
        ]
        existing_object_lookup_fields = [getattr(obj, lookup_field) for obj in existing_objects]
        non_existing_objects = [
            obj for obj in objs if getattr(obj, lookup_field) not in existing_object_lookup_fields
        ]
        self.bulk_create(non_existing_objects, batch_size=999)

        return super().get_queryset().filter(**lookup)

@dorosch
Copy link
Author

dorosch commented Nov 29, 2020

This manager seemed to not be working properly. Here is mine slightly modified version:

class BulkGetOrCreateManager(models.Manager):
    def bulk_get_or_create(self, objs, lookup_field=None):
        assert lookup_field, "Not set 'lookup_field' for 'bulk_get_or_create'"

        lookup = {f'{lookup_field}__in': [getattr(obj, lookup_field) for obj in objs]}
        existing_objects = [
            obj for obj in self.get_queryset().filter(**lookup)
        ]
        existing_object_lookup_fields = [getattr(obj, lookup_field) for obj in existing_objects]
        non_existing_objects = [
            obj for obj in objs if getattr(obj, lookup_field) not in existing_object_lookup_fields
        ]
        self.bulk_create(non_existing_objects, batch_size=999)

        return super().get_queryset().filter(**lookup)

It's look like true. Thanks you so much)

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment