ライブラリ

Django ORM

概要

Django ORMは、DjangoフレームワークのActive Record型ORMです。モデル駆動のデータベース設計、自動マイグレーション、管理インターフェースとの深い統合を特徴とし、PostgreSQL、MySQL、SQLite、Oracle等をサポートします。

詳細

Django ORMは、Convention over Configurationの思想に基づいて設計されており、最小限の設定でパワフルなデータベース操作を実現します。Pythonクラスとして定義されたモデルから自動的にデータベーステーブルを生成し、直感的なクエリAPIを提供します。Django管理インターフェースとの統合により、CRUD操作を簡単に実行できます。

主な特徴

• Active Recordパターン: モデルクラスに業務ロジックとデータアクセスを統合
• 自動マイグレーション: モデルの変更から自動的にマイグレーションファイルを生成
• 管理インターフェース: モデルの自動CRUD画面生成
• 豊富なフィールドタイプ: 多様なデータ型をサポート
• 強力なクエリセットAPI: 遅延評価による効率的なクエリ実行

メリット・デメリット

メリット

• 学習コストが低く、初心者にも優しい設計
• Django管理インターフェースによる圧倒的な開発生産性
• Python開発者にとって自然で読みやすいAPI
• 豊富なドキュメントとコミュニティサポート
• CRUD中心のWebアプリケーション開発に最適

デメリット

• 複雑なクエリ表現に制限がある場合がある
• Djangoフレームワーク依存のため他のフレームワークでは使用困難
• 大規模システムでのパフォーマンスチューニングが複雑
• ORM抽象化により生SQLの直接制御が困難

参考ページ

• Django ORM 公式ドキュメント
• Django モデルリファレンス
• Django GitHub

書き方の例

インストールと基本セットアップ

pip install Django
django-admin startproject myproject
cd myproject
python manage.py startapp myapp

# settings.py
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': 'mydatabase',
        'USER': 'myuser',
        'PASSWORD': 'mypassword',
        'HOST': 'localhost',
        'PORT': '5432',
    }
}

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'myapp',  # アプリケーションを追加
]

基本的なCRUD操作（モデル定義、作成、読み取り、更新、削除）

# models.py
from django.db import models

class User(models.Model):
    name = models.CharField(max_length=100)
    email = models.EmailField(unique=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    def __str__(self):
        return self.name

    class Meta:
        ordering = ['-created_at']

class Post(models.Model):
    title = models.CharField(max_length=200)
    content = models.TextField(blank=True)
    author = models.ForeignKey(User, on_delete=models.CASCADE, related_name='posts')
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    def __str__(self):
        return self.title

    class Meta:
        ordering = ['-created_at']

# マイグレーション実行
# python manage.py makemigrations
# python manage.py migrate

# 基本的なCRUD操作
from myapp.models import User, Post

# 作成
user = User.objects.create(
    name='田中太郎',
    email='[email protected]'
)

# 読み取り
all_users = User.objects.all()
user_by_email = User.objects.get(email='[email protected]')
filtered_users = User.objects.filter(name__icontains='田中')

# 更新
user = User.objects.get(id=1)
user.name = '田中次郎'
user.save()

# 一括更新
User.objects.filter(name__icontains='田中').update(name='更新済み')

# 削除
user = User.objects.get(id=1)
user.delete()

# 一括削除
User.objects.filter(name='テスト').delete()

高度なクエリとリレーションシップ

from django.db.models import Q, Count, Avg, Max, Min
from datetime import datetime, timedelta

# 複雑な条件クエリ
recent_users = User.objects.filter(
    Q(created_at__gte=datetime.now() - timedelta(days=30)) &
    Q(email__endswith='@company.com')
).order_by('-created_at')[:10]

# JOINクエリ（select_related）
posts_with_author = Post.objects.select_related('author').all()

# 逆参照クエリ（prefetch_related）
users_with_posts = User.objects.prefetch_related('posts').all()

# 集約クエリ
user_stats = User.objects.aggregate(
    total_users=Count('id'),
    avg_posts=Avg('posts__id'),
    latest_user=Max('created_at')
)

# グループ化クエリ
post_counts = User.objects.annotate(
    post_count=Count('posts')
).filter(post_count__gt=5)

# サブクエリ
active_users = User.objects.filter(
    id__in=Post.objects.values_list('author_id', flat=True).distinct()
)

# 複雑なクエリセット操作
complex_query = Post.objects.select_related('author').filter(
    author__email__endswith='@company.com',
    created_at__gte=datetime.now() - timedelta(days=7)
).annotate(
    author_post_count=Count('author__posts')
).order_by('-created_at', 'title')

# Raw SQLクエリ
users = User.objects.raw(
    "SELECT * FROM myapp_user WHERE email LIKE %s",
    ['%@company.com']
)

マイグレーションとスキーマ管理

# マイグレーションファイル生成
python manage.py makemigrations

# 特定アプリのマイグレーション
python manage.py makemigrations myapp

# マイグレーション実行
python manage.py migrate

# マイグレーション状況確認
python manage.py showmigrations

# マイグレーション SQL確認
python manage.py sqlmigrate myapp 0001

# カスタムマイグレーション
# migrations/0002_custom_migration.py
from django.db import migrations

def forwards_func(apps, schema_editor):
    User = apps.get_model('myapp', 'User')
    db_alias = schema_editor.connection.alias
    User.objects.using(db_alias).filter(
        name='old_name'
    ).update(name='new_name')

def reverse_func(apps, schema_editor):
    User = apps.get_model('myapp', 'User')
    db_alias = schema_editor.connection.alias
    User.objects.using(db_alias).filter(
        name='new_name'
    ).update(name='old_name')

class Migration(migrations.Migration):
    dependencies = [
        ('myapp', '0001_initial'),
    ]

    operations = [
        migrations.RunPython(forwards_func, reverse_func),
    ]

パフォーマンス最適化と高度な機能

from django.db import transaction
from django.core.cache import cache

# トランザクション管理
@transaction.atomic
def create_user_with_posts():
    user = User.objects.create(
        name='山田花子',
        email='[email protected]'
    )
    
    Post.objects.create(
        title='最初の投稿',
        content='Django ORMを使った投稿です',
        author=user
    )
    
    return user

# バッチ操作
users_data = [
    User(name='ユーザー1', email='[email protected]'),
    User(name='ユーザー2', email='[email protected]'),
    User(name='ユーザー3', email='[email protected]'),
]
User.objects.bulk_create(users_data)

# バッチ更新
User.objects.bulk_update(
    users_data,
    ['name', 'email']
)

# クエリセット最適化
# N+1問題を回避
efficient_query = Post.objects.select_related('author').prefetch_related(
    'author__posts'
).all()

# インデックスヒント
from django.db.models import Index

class User(models.Model):
    name = models.CharField(max_length=100)
    email = models.EmailField(unique=True)
    
    class Meta:
        indexes = [
            Index(fields=['name', 'email']),
            Index(fields=['-created_at']),
        ]

# カスタムマネージャー
class PublishedPostManager(models.Manager):
    def get_queryset(self):
        return super().get_queryset().filter(is_published=True)

class Post(models.Model):
    title = models.CharField(max_length=200)
    content = models.TextField()
    is_published = models.BooleanField(default=False)
    
    objects = models.Manager()  # デフォルトマネージャー
    published = PublishedPostManager()  # カスタムマネージャー

# 使用例
published_posts = Post.published.all()

フレームワーク統合と実用例

# Django REST Framework統合
from rest_framework import serializers, viewsets
from rest_framework.decorators import action
from rest_framework.response import Response

class UserSerializer(serializers.ModelSerializer):
    post_count = serializers.SerializerMethodField()
    
    class Meta:
        model = User
        fields = ['id', 'name', 'email', 'created_at', 'post_count']
    
    def get_post_count(self, obj):
        return obj.posts.count()

class UserViewSet(viewsets.ModelViewSet):
    queryset = User.objects.all()
    serializer_class = UserSerializer
    
    @action(detail=True, methods=['get'])
    def posts(self, request, pk=None):
        user = self.get_object()
        posts = user.posts.all()
        return Response([{
            'id': post.id,
            'title': post.title,
            'created_at': post.created_at
        } for post in posts])

# Celeryタスク統合
from celery import shared_task

@shared_task
def process_user_data(user_id):
    try:
        user = User.objects.get(id=user_id)
        # 重い処理をバックグラウンドで実行
        user.processed = True
        user.save()
        return f"User {user.name} processed successfully"
    except User.DoesNotExist:
        return f"User with id {user_id} not found"

# Django管理インターフェース
from django.contrib import admin

@admin.register(User)
class UserAdmin(admin.ModelAdmin):
    list_display = ['name', 'email', 'created_at', 'post_count']
    list_filter = ['created_at']
    search_fields = ['name', 'email']
    readonly_fields = ['created_at', 'updated_at']
    
    def post_count(self, obj):
        return obj.posts.count()
    post_count.short_description = '投稿数'

@admin.register(Post)
class PostAdmin(admin.ModelAdmin):
    list_display = ['title', 'author', 'created_at']
    list_filter = ['created_at', 'author']
    search_fields = ['title', 'content']
    autocomplete_fields = ['author']