読者です 読者をやめる 読者になる 読者になる

ORMを使わずに問い合わせてオブジェクトでごにょごにょ

Python Django

社内向け。検索用記事。Djangoのバージョンは1.4。
ORMを使わずにSQLを投げて、結果をオブジェクトとして使いたい場合。
bpcommonsは社内の人なら拾ってこれるはず。bpmappersはpipでインストール。

myapp/models.py

データベース用のものとふるまいをするだけのもので別にしてある。
実際はもっとカラムも多くてJOINを使うので複数のモデル(テーブルになる)

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


class User(models.Model):
    name = models.CharField(max_length=20)

    class Meta:
        db_table = 'user'


class ProfileA(models.Model):
    user = models.ForeignKey(User)
    age = models.IntegerField(db_index=True)

    class Meta:
        db_table = 'profile_a'


class ProfileB(models.Model):
    user = models.ForeignKey(User)
    point = models.IntegerField(default=0)

    class Meta:
        db_table = 'profile_b'


class Player(object):
    def __init__(self):
        self.age = None
        self.point = None

    def get_generation(self):
        if self.age:
            return u"%d代" % ((self.age / 10) * 10)
    generation = property(get_generation)

    def get_level(self):
        if self.point:
            return self.point / 100
    level = property(get_level)

Playerモデルはデータベースに関係ないので別のモジュールに作成してもいいかもしれない。

myapp/mappers.py

# coding: utf-8
from bpmappers import Mapper, RawField


class PlayerMapper(Mapper):
    Name = RawField('name')
    Generation = RawField('generation')
    Level = RawField('level')

myapp/api.py

ロジックコード。
get_players_list関数では、SQLを投げてuser、profile_a、profile_bテーブルを結合したものをPlayerモデルのインスタンスとして取得し、PlayerMapperで辞書にマッピングしている。

# coding: utf-8
from django.db import connection

from beproud.django.commons.utils.dbi import select

from myapp.models import Player
from myapp.mappers import PlayerMapper


def get_players_list():
    """すべてのプレイヤーのリストを返す
    """
    sql = """
  SELECT
    u.name,
    pa.age,
    pb.point
  FROM user u
  LEFT JOIN profile_a pa ON u.id = pa.user_id
  LEFT JOIN profile_b pb ON u.id = pb.user_id
"""
    result = select(connection, sql, model=Player)
    return [PlayerMapper(player).as_dict() for player in result]

この例のSQLならDjangoのORMでも問い合わせ可能だけど、実際はもっと複雑なSQL

myapp/tests.py

実行するのにとりあえずtests.pyを使う。
api.pyの関数の実行結果をpprintで画面に出すようにしている。

# coding: utf-8
from pprint import pprint

from django.test import TestCase

from myapp import api
from myapp.models import User, ProfileA, ProfileB


class PlayersListTest(TestCase):
    def setUp(self):
        for i in range(3):
            user = User.objects.create(name='Player%d' % i)
            profile_a = ProfileA.objects.create(
                user=user,
                age=(i + 1) * 10  # 10, 20, 30
            )
            profile_b = ProfileB.objects.create(
                user=user,
                point=(i + 1) * 200  # 200, 400, 600
            )

    def test_call(self):
        lst = api.get_players_list()
        pprint(lst)

テストの実行結果

(bpcommons_select)tokibito@ubuntu:~/bitbucket/sample_nullpobug/django/select_object$ python manage.py test myapp
Creating test database for alias 'default'...
[{'Generation': u'10\u4ee3', 'Name': u'Player0', 'Level': 2},
 {'Generation': u'20\u4ee3', 'Name': u'Player1', 'Level': 4},
 {'Generation': u'30\u4ee3', 'Name': u'Player2', 'Level': 6}]
.
----------------------------------------------------------------------
Ran 1 test in 0.010s

OK
Destroying test database for alias 'default'...

コードはbitbucketに。
tokibito / sample_nullpobug / source / django / select_object — Bitbucket