test_race_condition.py

#!/usr/bin/env python

import os
import logging

from threading import Thread

from django.core.management.base import BaseCommand, CommandError
from django.conf import settings

from optparse import OptionParser

# testing with postgresql 9.1
settings.configure(
    DATABASES={
        'default': {
            'ENGINE': 'django.db.backends.postgresql_psycopg2',
            'NAME': 'test',
            'USER': 'test',
            'PASSWORD': 's3cr3t',
            'HOST': '',
            'PORT': '',
            }
    },
    INSTALLED_APPS=[
        'django.contrib.auth',
    ],
)

from django.db import transaction
from django.contrib.auth.models import User

class MyThread(Thread):

    def __init__(self, username):
        super(MyThread, self).__init__(name=username)
        self.username = username

    def run(self):
        # passes with get_or_create
        instance, created = User.objects.get_or_create(username=self.username)

        # fails with try get .. except create
        #created = False
        #try:
            #instance = User.objects.get(username=self.username)
        #except User.DoesNotExist:
            #instance = User.objects.create(username=self.username)
            #created = False
        print '%s %s' % (instance.id, created)
        instance.delete()
        return

if __name__ == '__main__':
    parser = OptionParser()
    parser.add_option("--num-threads", type="int", default=5, dest="threads")
    options, args = parser.parse_args()
    for i in range(options.threads):
        thread = MyThread(username="test_user")
        thread.start()