def _update_or_create_shard(self, step):
        shard_index = random.randint(0, self.field.shard_count - 1)
        # Converting the set to a list introduces some randomness in the ordering, but that's fine
        shard_pks = list(self.field.value_from_object(self.instance)) # needs to be indexable
        try:
            shard_pk = shard_pks[shard_index]
        except IndexError:
            # We don't have this many shards yet, so create a new one
            with transaction.atomic(xg=True):
                # We must re-fetch the instance to ensure that we do this atomically, but we must
                # also update self.instance so that the calling code which is referencing
                # self.instance also gets the updated list of shard PKs
                new_shard = self._create_shard(count=step)
                new_instance = self.instance._default_manager.get(pk=self.instance.pk)
                new_instance_shard_pks = getattr(new_instance, self.field.attname, set())
                new_instance_shard_pks.add(new_shard.pk)
                setattr(self.instance, self.field.attname, new_instance_shard_pks)
                models.Model.save(new_instance) # avoid custom save method, which might do DB lookups
        else:
            with transaction.atomic():
                from djangae.models import CounterShard
                shard = CounterShard.objects.get(pk=shard_pk)
                shard.count += step
                shard.save()
Post a list of messages to a given Zaqar queue.
dataset/ETHPy150Open potatolondon/djangae/djangae/fields/counting.py/RelatedShardManager._update_or_create_shard--crossover with--dataset/ETHPy150Open openstack/rally/rally/plugins/openstack/scenarios/zaqar/utils.py/ZaqarScenario._messages_post