【django】UUIDFieldを使って、AutoFieldを置き換える

django1.8から追加された、UUIDFieldの使い方を紹介します。このクラスを使うと、UUIDを扱えるフィールドをモデルに追加できます。

UUIDFieldの用途としては、djangoのドキュメントに書いてある通り、プライマリキーにするAutoFieldの代わりになります。例えばランキングテーブルのような、大量のデータを高頻度に削除し追加する場合など、AutoFieldを使いたくない場合にUUIDFieldを使うことができます。

UUIDFieldの基本的な使い方

ドキュメントに書いてある通りですが、UUIDFieldは以下のように使うことができます。

モデルの作成

UUIDFieldを持ったFoo1クラスを、myapp/models.pyに作成します。そのUUIDFieldのdefaultで、値がランダムなUUIDになるようにしています。

# coding: utf-8
import uuid
from django.db import models


class Foo1(models.Model):
    id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
    name = models.CharField(max_length=64)

データ登録などを確認

./manage.py shellを使って、レコードを登録などを確認してみます。

$ ./manage.py shell

# レコードを登録します。
>>> from myapp.models import Foo1
>>> foo = Foo1()
>>> foo.name = 'foo1'
>>> foo.save()

# レコードが登録されたことを確認します。
>>> Foo1.objects.count()
1

# idはUUIDクラスのオブジェクトになっています。
>>> foo.id
UUID('3c89f781-ef2f-4d9f-9b21-dc0896e53b49')
>>> type(foo.id)
<class 'uuid.UUID'>

# UUID.hexで、32文字の16進数文字列で取得できます。
>>> foo.id.hex
'3c89f781ef2f4d9f9b21dc0896e53b49'

カラムの型を確認するとSQLiteの場合はchar(32)になっていました。また、idは32桁の16進数文字列になっています。

$ sqlite3 db.sqlite3
sqlite> .schema myapp_foo1
CREATE TABLE "myapp_foo1" ("name" varchar(64) NOT NULL, "id" char(32) NOT NULL PRIMARY KEY);
sqlite> SELECT id FROM myapp_foo1;
3c89f781ef2f4d9f9b21dc0896e53b49

UUIDFieldの重複を考慮する

上記Foo1のように、「default=uuid.uuid4」としていると、ランダムで決定したUUIDに重複が発生した場合が気になります。その場合に例外を発生させないように、重複発生時にはランダムのUUIDを再生性し、登録を試みるように変更してみます。

retried_insertの作成

myapp/utils.pyを作成し、retried_insertという関数を作成しました。この関数では重複エラーが発生した場合に、再度UUIDをランダムで生成するようにしています。

また、再度UUIDを作る処理が無限ループに陥らないように、9回再生性(計10回)し、それでも一意なUUIDを作れなかった場合は例外を投げるようにしています。

後はテスト用に、引数pkdictを用意しています。この引数に値が渡されると、ランダムのUUIDではなく、pkdictに入ったUUIDを使うようになります。それによって、わざと重複エラーを発生させることができるので、テストケースが書きやすくなります。

# coding: utf-8
import uuid
import time
from django.db import transaction
from django.db.utils import IntegrityError


def retried_insert(instance, pkdict={}):
    """
    インスタンスの内容をDBに登録します。
    もし、PK重複により登録失敗したら、PKを再生成して登録を試みます。
    Args:
        instance: 登録対象のインスタンス
        pkdict: PKのディクショナリ
            これはテスト時に、重複エラーを発生させるために用意してあります。
    """
    max_retries = 9  # 最大再試行回数
    for idx in range(max_retries + 1):
        try:
            with transaction.atomic(savepoint=True):
                instance.pk = pkdict.get(idx) or uuid.uuid4()
                instance.save(force_insert=True)
                break
        except IntegrityError:
            time.sleep(0.1)
    else:
        # インスタンスの登録ができなかった場合は例外を投げます。
        raise IntegrityError('DBに登録できませんでした。[%s].' % type(instance))

モデルの作成

retried_insertを使って登録する、Foo2クラスを作成します。

class Foo2(models.Model):
    id = models.UUIDField(primary_key=True)
    name = models.CharField(max_length=64)

retried_insertを試す

./manage.py shellを使って、retried_insertを確認してみます。

>>> from myapp.models import Foo2
>>> from myapp.utils import retried_insert
>>> foo = Foo2()
>>> foo.name = 'foo2'

# retried_insertを試します。
>>> retried_insert(foo)
>>> foo.id
UUID('aea9f710-b8bf-4acd-bc2e-ba0c82b5e384')
>>> foo.id.hex
'aea9f710b8bf4acdbc2eba0c82b5e384'

テストケース

myapp/tests.pyにテストケースを作成します。テスト内容は、

  • 重複発生なしに登録成功するケース
  • 一意なPKを作れずに失敗するケース
  • 数回失敗後、一意なPKを作れて登録成功するケース

の3種類です。

# coding: utf-8

import uuid
from django.test import TestCase
from django.db.utils import IntegrityError
from myapp.models import Foo2
from myapp.utils import retried_insert


class TestRetriedInsert(TestCase):
    def test_retried_insert_success(self):
        """
        重複発生なしに登録成功するケース
        """
        # データを準備します。
        foo = Foo2()
        foo.name = 'foo2'

        # テスト対象を実行します。
        retried_insert(foo)

        # テスト結果を確認します。
        self.assertEqual(type(foo.pk), uuid.UUID)
        self.assertEqual(Foo2.objects.count(), 1)

    def test_retried_insert_failed(self):
        """
        一意なPKを作れずに失敗するケース
        """
        # データを準備します。
        foo = Foo2()
        foo.name = 'foo2'

        # データを登録しておきます。
        retried_insert(foo)
        self.assertEqual(Foo2.objects.count(), 1)

        # PKが入ったディクショナリを作成します。
        pk = foo.pk
        pkdict = {idx: pk for idx in range(10)}

        # テスト対象を実行し、IntegrityErrorを投げることを確認します。
        self.assertRaises(IntegrityError, retried_insert, foo, pkdict)

        # データが追加されていないことを確認します。
        self.assertEqual(Foo2.objects.count(), 1)

    def test_retried_insert_failed_success(self):
        """
        数回失敗後、一意なPKを作れて登録成功するケース
        """
        # データを準備します。
        foo = Foo2()
        foo.name = 'foo2'

        # データを登録しておきます。
        retried_insert(foo)
        self.assertEqual(Foo2.objects.count(), 1)

        # PKが入ったディクショナリを作成します。
        # ※5回までは重複を発生させ、それ以降はランダムなUUIDを生成するようにします。
        pk = foo.pk
        pkdict = {idx: pk for idx in range(5)}

        # テスト対象を実行します。
        retried_insert(foo, pkdict)

        # データが追加されていることを確認します。
        self.assertEqual(Foo2.objects.count(), 2)

この記事が役に立った場合、シェアしていただけると励みになります!!