3.

Django でのデータベース接続と操作完全ガイド — ORM/QuerySet/トランザクション

編集
この記事の要点
  • Django の DB 設定は settings.pyDATABASES 辞書。 SQLite / PostgreSQL / MySQL / Oracle に対応
  • スキーマは Model クラスから自動生成: makemigrations でマイグレーションファイル作成 → migrate で DB 反映
  • クエリは QuerySet: Model.objects.filter(...).order_by(...).first()。 遅延評価で必要時にのみ SQL 発行
  • 生 SQL 必要時は Model.objects.raw("SELECT ...") または connection.cursor()
  • トランザクションは transaction.atomic() ブロックで囲む。 ネスト可、 セーブポイント自動

DATABASES 設定

# config/settings.py

# SQLite (デフォルト、 開発用)
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.sqlite3',
        'NAME': BASE_DIR / 'db.sqlite3',
    }
}

# PostgreSQL
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': 'mydb',
        'USER': 'postgres',
        'PASSWORD': 'pass',
        'HOST': '127.0.0.1',
        'PORT': '5432',
        'CONN_MAX_AGE': 60,         # 接続プール (秒)
        'OPTIONS': {
            'sslmode': 'require',
        },
    }
}

# MySQL
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.mysql',
        'NAME': 'mydb',
        'USER': 'root',
        'PASSWORD': 'pass',
        'HOST': '127.0.0.1',
        'PORT': '3306',
        'OPTIONS': {
            'charset': 'utf8mb4',
            'init_command': "SET sql_mode='STRICT_TRANS_TABLES'",
        },
    }
}

# 環境変数から(推奨)
import environ
env = environ.Env()
DATABASES = {
    'default': env.db('DATABASE_URL'),
    # postgres://user:pass@host:5432/dbname
}

マイグレーション

# モデル変更 → マイグレーションファイル作成
python manage.py makemigrations

# 出力例
# Migrations for 'blog':
#   blog/migrations/0001_initial.py
#     - Create model Article

# マイグレーション適用
python manage.py migrate

# 特定アプリのみ
python manage.py migrate blog
python manage.py migrate blog 0003   # 特定リビジョンへ

# マイグレーション一覧
python manage.py showmigrations

# 中身を SQL で確認
python manage.py sqlmigrate blog 0001

# ロールバック(一つ前へ)
python manage.py migrate blog 0002
# 完全ロールバック
python manage.py migrate blog zero

# マイグレーションファイル削除(要注意)
# blog/migrations/0001_initial.py を消すとローカルでは作り直せるが、
# 既に migrate 済の環境では django_migrations テーブルと不整合

Model 定義

# blog/models.py
from django.db import models
from django.contrib.auth.models import User

class Category(models.Model):
    name = models.CharField(max_length=50, unique=True)
    slug = models.SlugField(unique=True)

    class Meta:
        verbose_name_plural = "Categories"

    def __str__(self):
        return self.name

class Article(models.Model):
    STATUS_CHOICES = [
        ('draft', '下書き'),
        ('published', '公開'),
    ]

    title = models.CharField(max_length=200)
    slug = models.SlugField(unique=True)
    body = models.TextField()
    category = models.ForeignKey(
        Category,
        on_delete=models.CASCADE,
        related_name='articles',
    )
    author = models.ForeignKey(User, on_delete=models.PROTECT)
    tags = models.ManyToManyField('Tag', blank=True)

    status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='draft')
    views = models.PositiveIntegerField(default=0)
    published_at = models.DateTimeField(null=True, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

    class Meta:
        ordering = ['-created_at']
        indexes = [
            models.Index(fields=['status', 'published_at']),
        ]
        constraints = [
            models.UniqueConstraint(
                fields=['author', 'title'],
                name='unique_author_title',
            ),
        ]

    def __str__(self):
        return self.title

QuerySet (CRUD)

from blog.models import Article, Category
from django.contrib.auth.models import User

# === Create ===
article = Article.objects.create(
    title="Django 入門",
    slug="django-intro",
    body="...",
    author=User.objects.get(username='taro'),
    category=Category.objects.get(slug='tech'),
)

# あるいは
a = Article(title="...", body="...")
a.save()

# === Read ===
# 1件取得
art = Article.objects.get(pk=1)            # 存在しないと DoesNotExist
art = Article.objects.filter(pk=1).first() # 存在しないと None
art = Article.objects.get_or_create(slug='intro', defaults={'title': '...'})

# 全件
all_articles = Article.objects.all()

# フィルタ
published = Article.objects.filter(status='published')
recent = Article.objects.filter(created_at__gte='2026-01-01')
search = Article.objects.filter(title__icontains='Django')

# 否定・OR
from django.db.models import Q
Article.objects.filter(~Q(status='draft'))
Article.objects.filter(Q(category__slug='tech') | Q(category__slug='news'))

# ソート・ページング
Article.objects.order_by('-created_at')[:10]    # 最新 10 件

# 関連先 (ForeignKey)
art = Article.objects.select_related('author', 'category').get(pk=1)
art.author.username
art.category.name

# 関連先 (ManyToMany / 逆参照)
arts = Article.objects.prefetch_related('tags').all()
for art in arts:
    print([t.name for t in art.tags.all()])

# 集約
from django.db.models import Count, Avg, Sum, Max
Category.objects.annotate(article_count=Count('articles'))
Article.objects.aggregate(total_views=Sum('views'))

# === Update ===
art = Article.objects.get(pk=1)
art.title = "新タイトル"
art.save()

# 一括更新(save() を呼ばない、 シグナルも飛ばない)
Article.objects.filter(status='draft').update(status='published')

# === Delete ===
art = Article.objects.get(pk=1)
art.delete()

# 一括削除
Article.objects.filter(status='draft').delete()

QuerySet ルックアップ(フィールド検索)

ルックアップ意味
exact完全一致 (デフォルト)filter(title__exact='Hello')
iexact大文字小文字無視filter(slug__iexact='HELLO')
contains / icontains部分一致filter(title__icontains='django')
startswith / endswith前方/後方filter(slug__startswith='2026-')
inリスト内filter(id__in=[1, 2, 3])
gt / gte / lt / lte比較filter(views__gte=100)
range範囲filter(created_at__range=(d1, d2))
isnullNULL チェックfilter(published_at__isnull=True)
year / month / day日付パーツfilter(created_at__year=2026)

生 SQL を実行する

# 1. Model.objects.raw() — Model インスタンスを返す
articles = Article.objects.raw(
    'SELECT * FROM blog_article WHERE views > %s ORDER BY views DESC',
    [100]
)
for art in articles:
    print(art.title)

# 2. connection.cursor() — 完全な生クエリ
from django.db import connection

with connection.cursor() as cursor:
    cursor.execute("""
        SELECT category_id, COUNT(*) AS c
        FROM blog_article
        WHERE status = %s
        GROUP BY category_id
    """, ['published'])

    for row in cursor.fetchall():
        print(row)

# 辞書で取得
def dictfetchall(cursor):
    columns = [col[0] for col in cursor.description]
    return [dict(zip(columns, row)) for row in cursor.fetchall()]

with connection.cursor() as cursor:
    cursor.execute("SELECT id, title FROM blog_article")
    rows = dictfetchall(cursor)

トランザクション

from django.db import transaction, IntegrityError

# 1. デコレータ
@transaction.atomic
def transfer(from_acc, to_acc, amount):
    from_acc.balance -= amount
    from_acc.save()
    to_acc.balance += amount
    to_acc.save()

# 2. with 文
def create_order(user, items):
    with transaction.atomic():
        order = Order.objects.create(user=user)
        for item in items:
            OrderItem.objects.create(order=order, product=item)
        # ここで例外が出れば全部ロールバック

# 3. ネスト(セーブポイント)
def complex_op():
    with transaction.atomic():           # 外側
        order = Order.objects.create()
        try:
            with transaction.atomic():   # 内側(セーブポイント)
                send_email_log(order)
        except SMTPException:
            # 内側だけロールバック、 外側は継続
            pass

# 4. select_for_update (行ロック)
with transaction.atomic():
    account = Account.objects.select_for_update().get(pk=acc_id)
    account.balance -= amount
    account.save()

# 5. autocommit 切替(必要時のみ)
from django.db import transaction
transaction.set_autocommit(False)
try:
    # 手動制御
    Article.objects.create(...)
    transaction.commit()
except:
    transaction.rollback()
finally:
    transaction.set_autocommit(True)

パフォーマンスのコツ

  • N+1 問題対策: select_related(FK は JOIN)、 prefetch_related(M2M / 逆 FK は別クエリ)
  • 必要な列のみ: .only('id', 'title') / .defer('body')
  • カウントは .count(): len(qs) は全件評価する
  • 存在チェックは .exists(): if qs: は全件評価する
  • 大量更新は .update(): 1 件ずつ save() しない
  • bulk_create / bulk_update: 大量挿入
  • インデックス: Meta.indexes で複合インデックス定義
  • EXPLAIN: qs.explain() で SQL プラン確認

FAQ

Q: migrations フォルダは git に入れる?
A: 入れます。 チーム全体で同じ DB スキーマを保証するため必須。

Q: スキーマを大幅変更したい
A: 開発初期なら全マイグレーション削除 + DB 削除 + makemigrations 再実行。 運用後はカスタムマイグレーションで段階的に。

Q: 複数 DB を使い分けたい
A: DATABASES に複数定義 → using('replica') で切替、 または DATABASE_ROUTERS で自動振り分け。

編集
Post Share
子ページ
  1. MySQL/MariaDBへの接続
  2. sqliteへの接続
  3. SELECT, INSERT, UPDATE, DELETE
  4. 素のSQLを直接実行する方法
  5. Order by DESCの指定方法
  6. limit, offsetの指定方法
  7. filterの検索オプション
  8. django-filterのlookup_expr検索オプション
  9. モデルの内部結合(1対1)
同階層のページ
  1. 環境構築とプロジェクト/アプリの作成
  2. MVC(MVT)のそれぞれの使い方と説明
  3. データベースへの接続と操作
  4. Django Administration
  5. git管理
  6. エラー一覧
  7. バージョンの確認方法
  8. ログ出力方法
  9. SQLのログ出力方法
  10. ログのローテート設定
  11. settings.pyの定数にアクセスする方法
  12. 本番環境へのインストールとアプリのデプロイ(apache編)
  13. 本番環境へのインストールとアプリのデプロイ(nginx編)
  14. djangoアプリの本番の開始URLを変更する
  15. 静的(static)ファイルの置き場所と読み込み(画像、css、js )
  16. CSRFトークンをAjaxで使用する方法
  17. ajaxの使用例(POST編)
  18. ファイルのアップロードとファイルの名前
  19. クイックスタート/チュートリアル
  20. ログイン機能
  21. テンプレート側のログイン判定
  22. ビュー側のログイン判定
  23. 管理者ユーザーの作成/判定と管理画面
  24. モデルのjson化とレスポンス
  25. runserverでポートを指定する方法
  26. cronによるバッチ実行
  27. テンプレートで利用する共通のcontextを定義する方法
  28. プログラムが本番サーバーで反映されない場合の対処法
  29. APIの作成
  30. cron用コマンド・ファイルの作成