Flaskデータベース連携（SQLAlchemy）

2026-02-04

はじめに

Webアプリケーション開発においてデータの保存と管理は不可欠な要素です。Flaskだけではデータを永続的に保存することはできません。そこでデータベースとの連携が必要となります。しかし生のSQLを直接扱うことは煩雑で、セキュリティリスクも伴います。この問題を解決するのがORMという技術です。本記事ではFlaskでデータベースを扱うための標準的なライブラリであるSQLAlchemyの基本について、初心者の方にもわかりやすく解説していきます。ORMの概念からモデルの定義方法、実際のデータ操作まで、具体的なコード例を交えながら順を追って説明します。

ORMの概念理解

ORMはObject-Relational Mappingの略称です。日本語ではオブジェクト関係マッピングと訳されます。これはプログラミングのオブジェクトとデータベースのテーブルを対応づける技術を指します。

従来のデータベース操作ではSQL文を直接記述する必要がありました。しかしSQL文は文字列として扱われるため、構文エラーに気づきにくい、タイプミスによるバグが発生しやすいといった問題がありました。またアプリケーションのコードとデータベースの構造が密接に結びついてしまうため、データベースを変更するとアプリケーション側も大きく修正する必要が出てくる場合がありました。

ORMを利用すると、Pythonのクラスとメソッドを使ってデータベース操作を行うことができます。例えばユーザーデータを扱う場合、Userというクラスを定義し、そのクラスのインスタンスを作成して保存するだけでデータベースにレコードが追加されます。データの取得も同様に、クラスメソッドを呼び出すことでオブジェクトとして結果を得ることができます。

このようにORMを使用することで、開発者はデータベースの種類（SQLite、MySQL、PostgreSQLなど）に依存せず、一貫した方法でデータ操作を行えるようになります。またSQLインジェクションのようなセキュリティリスクも軽減されます。Pythonの世界ではSQLAlchemyが最も広く使用されているORMの一つで、Flaskとの連携も非常にスムーズです。

モデル定義の基本

SQLAlchemyを使用するには、まずFlaskアプリケーションで設定を行う必要があります。初期設定ではデータベースのURIを指定し、SQLAlchemyオブジェクトを作成します。

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///example.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

db = SQLAlchemy(app)

次にデータベースのテーブルに対応するモデルクラスを定義します。モデルクラスはdb.Modelを継承して作成します。各クラス属性はデータベースのカラムに対応し、db.Columnを使用して定義します。

例えばブログアプリケーションの投稿を管理するPostモデルを考えてみましょう。

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100), nullable=False)
    content = db.Column(db.Text, nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

このモデル定義ではid、title、content、created_atの4つのカラムを持っています。idは整数型の主キー、titleは最大100文字の文字列で必須項目、contentはテキスト型で必須項目、created_atは日時型でデフォルト値として現在時刻が設定されます。

モデルを定義したら、データベーステーブルを作成する必要があります。これは以下のコードで行えます。

with app.app_context():
    db.create_all()

この処理により、定義したモデルに対応するテーブルがデータベース内に作成されます。既存のデータベースがある場合は、モデルに合わせてテーブルが更新されるわけではないことに注意してください。データベースの変更管理にはAlembicなどのマイグレーションツールを使用することをお勧めします。

モデル定義時には、データ型の他にも様々なオプションを指定できます。例えばユニーク制約、インデックスの設定、デフォルト値の指定などが可能です。これらのオプションを適切に使用することで、データの整合性を保ち、パフォーマンスを向上させることができます。

簡単なCRUD操作

データベース操作の基本はCRUDです。これはCreate（作成）、Read（読み取り）、Update（更新）、Delete（削除）の頭文字を取った用語です。SQLAlchemyを使用すると、これらの操作をPythonのオブジェクト指向的な方法で実行できます。

まずCreate操作から見ていきましょう。新しいレコードを作成するには、モデルクラスのインスタンスを作成し、データベースセッションに追加してコミットします。

# 新しい投稿を作成
new_post = Post(title='初めての投稿', content='これは初めての投稿内容です。')
db.session.add(new_post)
db.session.commit()

このコードではPostクラスのインスタンスを作成し、titleとcontentに値を設定しています。idとcreated_atは自動的に設定されるため、明示的に指定する必要はありません。db.session.add()でセッションにオブジェクトを追加し、db.session.commit()で変更をデータベースに反映させます。

次にRead操作について説明します。データの取得にはいくつかの方法があります。すべてのレコードを取得する場合は以下のようにします。

# すべての投稿を取得
all_posts = Post.query.all()

特定の条件に合致するレコードを取得する場合はfilter_byやfilterを使用します。

# タイトルが「初めての投稿」である投稿を取得
post = Post.query.filter_by(title='初めての投稿').first()

# より複雑な条件での検索
recent_posts = Post.query.filter(Post.created_at >= datetime(2023, 1, 1)).all()

主キーによる検索はより簡単です。

# IDが1の投稿を取得
post = Post.query.get(1)

Update操作は、既存のオブジェクトの属性を変更し、セッションをコミットするだけで完了します。

# 投稿を更新
post = Post.query.get(1)
post.title = '更新されたタイトル'
post.content = '更新された内容'
db.session.commit()

最後にDelete操作です。オブジェクトを削除するには、セッションから削除を指示し、コミットします。

# 投稿を削除
post = Post.query.get(1)
db.session.delete(post)
db.session.commit()

これらの基本的なCRUD操作を組み合わせることで、ほとんどのデータベース操作を行うことができます。SQLAlchemyはさらに高度なクエリ機能も提供しており、テーブル結合、集計関数、トランザクション管理など、複雑なデータ操作も可能にしています。

実際のアプリケーション開発では、これらの操作をFlaskのルート関数内で使用します。例えばブログ投稿の一覧表示、新規作成、編集、削除などの機能を実装する際に、上記のCRUD操作を応用します。

まとめ

本記事ではFlaskとSQLAlchemyを使用したデータベース連携の基本について解説しました。ORMの概念から始め、モデルの定義方法、そして実際のCRUD操作について具体的なコード例を交えて説明しました。

ORMを利用することで、データベース操作をオブジェクト指向のパラダイムで扱えるようになります。これによりコードの可読性が向上し、保守性も高まります。またデータベースの種類に依存しない抽象化されたインターフェースを提供するため、アプリケーションの移植性も向上します。

モデル定義では、データベースのテーブル構造をPythonのクラスとして表現します。各カラムはクラス属性として定義し、データ型や制約を指定します。これによりデータの整合性を保ちながら、直感的なデータ操作が可能になります。

CRUD操作はデータベース利用の基本です。SQLAlchemyではこれらの操作をPythonのメソッドチェーンやオブジェクト操作として表現できます。生のSQLを書く必要がほとんどなく、安全かつ効率的にデータベースを操作できます。

演習問題

初級問題（3問）

初級1: モデル定義

ユーザー情報を管理するUserモデルを定義してください。
【要件】

  • id: 整数型、主キー
  • username: 文字列型(50文字以内)、必須、ユニーク
  • email: 文字列型(120文字以内)、必須、ユニーク
  • created_at: 日時型、デフォルトで現在時刻

初級2: データ作成

Userモデルを使用して、新しいユーザーをデータベースに追加するコードを書いてください。
【ユーザー情報】

  • username: “tanaka_taro”
  • email: “taro@example.com”

初級3: データ検索

ユーザー名が”tanaka_taro”であるユーザーを検索し、そのユーザーのemailを表示するコードを書いてください。

中級問題（6問）

中級1: リレーションシップ

UserモデルとPostモデルを1対多の関係で定義してください。
【Postモデルの要件】

  • id: 整数型、主キー
  • title: 文字列型(100文字以内)、必須
  • content: テキスト型、必須
  • user_id: 整数型、外部キー（Userモデルのidを参照）
  • created_at: 日時型、デフォルトで現在時刻

中級2: リレーションシップを使ったデータ作成

ユーザーとそのユーザーに属する投稿を同時に作成するコードを書いてください。
【データ】

  • ユーザー: username=”sato_hanako”, email=”hanako@example.com”
  • 投稿: title=”はじめまして”, content=”初めての投稿です”

中級3: 条件付き検索

2024年1月1日以降に作成された投稿をすべて取得するクエリを書いてください。

中級4: データ更新

ユーザー名が”sato_hanako”のユーザーのemailを”hanako_sato@example.com”に更新するコードを書いてください。

中級5: 集計クエリ

各ユーザーが作成した投稿数を集計するクエリを書いてください。

中級6: トランザクション処理

ユーザーと投稿の作成をトランザクション内で行い、いずれかが失敗した場合はロールバックするコードを書いてください。

上級問題（3問）

上級1: 複雑な検索とソート

投稿タイトルに「Flask」を含み、2024年中に作成された投稿を、作成日時の新しい順に取得するクエリを書いてください。また、各投稿のユーザー名も一緒に取得してください。

上級2: バリデーション付きモデル

Userモデルに以下のバリデーションを追加してください。

  • username: 3文字以上20文字以下、英数字とアンダースコアのみ許可
  • email: 有効なメールアドレス形式であること

上級3: ソフトデリートの実装

Postモデルにソフトデリート機能を実装してください。
【要件】

  • deleted_atカラムを追加（日時型、NULL許可）
  • デフォルトのクエリではdeleted_atがNULLのレコードのみ取得
  • 削除時は実際に削除せず、deleted_atに現在時刻を設定
  • 削除されたレコードも取得できる方法を提供

解答例

初級問題解答例（3問）

初級1 解答例

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(50), nullable=False, unique=True)
    email = db.Column(db.String(120), nullable=False, unique=True)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

初級2 解答例

new_user = User(username="tanaka_taro", email="taro@example.com")
db.session.add(new_user)
db.session.commit()

初級3 解答例

user = User.query.filter_by(username="tanaka_taro").first()
if user:
    print(user.email)

中級問題解答例（6問）

中級1 解答例

class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100), nullable=False)
    content = db.Column(db.Text, nullable=False)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    user = db.relationship('User', backref=db.backref('posts', lazy=True))

中級2 解答例

user = User(username="sato_hanako", email="hanako@example.com")
post = Post(title="はじめまして", content="初めての投稿です", user=user)
db.session.add(user)
db.session.add(post)
db.session.commit()

中級3 解答例

from datetime import datetime
posts = Post.query.filter(Post.created_at >= datetime(2024, 1, 1)).all()

中級4 解答例

user = User.query.filter_by(username="sato_hanako").first()
if user:
    user.email = "hanako_sato@example.com"
    db.session.commit()

中級5 解答例

from sqlalchemy import func
result = db.session.query(
    User.username, 
    func.count(Post.id)
).join(Post).group_by(User.id).all()

中級6 解答例

try:
    user = User(username="test_user", email="test@example.com")
    post = Post(title="テスト", content="テスト投稿", user=user)
    db.session.add(user)
    db.session.add(post)
    db.session.commit()
except Exception as e:
    db.session.rollback()
    print(f"エラーが発生しました: {e}")

上級問題解答例（3問）

上級1 解答例

from sqlalchemy import and_, or_
posts = Post.query.join(User).filter(
    and_(
        Post.title.contains('Flask'),
        Post.created_at >= datetime(2024, 1, 1),
        Post.created_at < datetime(2025, 1, 1)
    )
).order_by(Post.created_at.desc()).all()

上級2 解答例

import re
from wtforms.validators import ValidationError

class User(db.Model):
    # ... 既存のカラム定義 ...

    def validate_username(self, username):
        if len(username) < 3 or len(username) > 20:
            raise ValidationError('ユーザー名は3文字以上20文字以下でなければなりません')
        if not re.match(r'^\w+$', username):
            raise ValidationError('ユーザー名は英数字とアンダースコアのみ使用できます')

    def validate_email(self, email):
        if not re.match(r'^[^@]+@[^@]+\.[^@]+$', email):
            raise ValidationError('有効なメールアドレスを入力してください')

上級3 解答例

class Post(db.Model):
    # ... 既存のカラム定義 ...
    deleted_at = db.Column(db.DateTime, nullable=True)

    @classmethod
    def query_active(cls):
        return cls.query.filter_by(deleted_at=None)

    @classmethod
    def query_with_deleted(cls):
        return cls.query

    def soft_delete(self):
        self.deleted_at = datetime.utcnow()
        db.session.commit()

# 使用例
# アクティブな投稿のみ取得
active_posts = Post.query_active().all()

# 削除済みを含むすべての投稿を取得
all_posts = Post.query_with_deleted().all()

# ソフトデリート実行
post.soft_delete()

同じカテゴリーの記事