Flask + PostgreSQL チュートリアル：データベース接続方法

By /

はい、承知いたしました。FlaskとPostgreSQLを連携させる方法について、データベース接続に焦点を当てた詳細なチュートリアル記事を作成します。約5000語のボリュームを目指し、各ステップ、概念、および関連事項を深く掘り下げて説明します。

Flask + PostgreSQL チュートリアル：データベース接続方法の詳細解説

はじめに

Webアプリケーション開発において、データベースは不可欠な要素です。ユーザーデータ、設定情報、コンテンツなど、アプリケーションが扱うほとんどの情報を永続的に保存するために使用されます。PythonのマイクロWebフレームワークであるFlaskは、そのシンプルさと柔軟性から多くの開発者に利用されています。そして、堅牢性、信頼性、および豊富な機能を備えたオープンソースのリレーショナルデータベースであるPostgreSQLは、多くのWebアプリケーションのバックエンドとして選択されています。

FlaskとPostgreSQLを組み合わせることで、強力でスケーラブルなWebアプリケーションを構築できます。本記事では、FlaskアプリケーションからPostgreSQLデータベースに接続し、基本的なデータベース操作を行う方法について、その詳細を網羅的に解説します。単にコードを示すだけでなく、なぜそのようにするのか、どのような選択肢があるのか、そして実際の開発で考慮すべき点（エラーハンドリング、セキュリティ、パフォーマンスなど）についても深く掘り下げていきます。

対象読者:

  • Flaskを使ったWebアプリケーション開発の基礎を理解している方
  • リレーショナルデータベース（特にPostgreSQL）の基本的な概念を知っている方
  • FlaskアプリケーションとPostgreSQLを連携させたいと考えている方

記事のゴール:

本記事を読み終えることで、以下のことができるようになることを目指します。

  • FlaskアプリケーションからPostgreSQLに接続するための開発環境を構築できる。
  • Pythonのライブラリ（psycopg2, Flask-SQLAlchemy）を使ってPostgreSQLに接続し、基本的な操作を実行できる。
  • ORM（SQLAlchemy）の概念を理解し、Flask-SQLAlchemyを使ってデータベースを操作するメリットを理解できる。
  • Flask-SQLAlchemyを使ったモデル定義、CRUD操作、セッション管理の方法を習得できる。
  • データベース接続に関連するエラーハンドリングやセキュリティ、パフォーマンスに関する基本的な知識を身につける。

今回は、簡単なタスク管理アプリケーションを例に、データベース接続と操作の実装方法を具体的に示していきます。

開発環境の準備

まず、チュートリアルを進めるために必要な開発環境を準備します。以下のソフトウェアが必要になります。

  1. Python 3.6以上: Flaskと関連ライブラリを実行するために必要です。
  2. Flask: Webアプリケーションフレームワークです。
  3. PostgreSQL: データベースサーバーです。
  4. Python用PostgreSQLアダプター (psycopg2 または psycopg2-binary): PythonからPostgreSQLに接続するためのライブラリです。
  5. ORMライブラリ (Flask-SQLAlchemy): データベース操作をより簡単に、オブジェクト指向的に行うためのライブラリです。

1. Pythonのインストール

お使いのOS（Windows, macOS, Linux）に合わせて、公式サイトからPythonの最新バージョンをダウンロードしてインストールしてください。インストール後、ターミナルやコマンドプロンプトで以下のコマンドを実行し、Pythonが正しくインストールされ、パスが通っているか確認します。

“`bash
python –version

または

python3 –version
“`

2. 仮想環境の利用推奨

プロジェクトごとに独立したPython環境を構築するために、仮想環境（venvやvirtualenv）を利用することを強く推奨します。これにより、プロジェクト間でライブラリのバージョン衝突を防ぐことができます。

プロジェクトディレクトリを作成し、その中で仮想環境を構築・有効化します。

“`bash
mkdir flask-postgres-tutorial
cd flask-postgres-tutorial

仮想環境の作成 (Python 3.3以上の場合)

python3 -m venv venv

仮想環境の有効化

macOS / Linux:

source venv/bin/activate

Windows:

venv\Scripts\activate

“`

仮想環境が有効化されると、プロンプトの先頭に(venv)のような仮想環境名が表示されます。

3. Flaskと必要なPythonライブラリのインストール

仮想環境が有効な状態で、必要なライブラリをpipを使ってインストールします。

bash
pip install Flask psycopg2-binary Flask-SQLAlchemy

  • Flask: Flaskフレームワーク本体です。
  • psycopg2-binary: PostgreSQLに接続するためのPythonアダプターです。psycopg2 と psycopg2-binary がありますが、バイナリ版の方が通常はコンパイルなしでインストールできるため簡単です。本番環境ではlibpqなどのCライブラリに依存するpsycopg2を使うことが多いですが、開発環境ではpsycopg2-binaryで十分です。
  • Flask-SQLAlchemy: FlaskとSQLAlchemyを連携させるための拡張機能です。

4. PostgreSQLのインストールと設定

お使いのOSに合わせてPostgreSQLをインストールします。macOSの場合はHomebrew、Windowsの場合は公式インストーラー、Linuxの場合はディストリビューションのパッケージマネージャー（apt, yumなど）を利用するのが一般的です。

インストール後の基本的な設定:

通常、PostgreSQLのインストール時に、デフォルトのスーパーユーザーであるpostgresが作成されます。セキュリティのため、専用のユーザーとデータベースを作成して使用することを推奨します。

PostgreSQLサーバーが起動していることを確認した後、psqlコマンドラインツールを使って設定を行います。

“`bash

デフォルトユーザー (postgres) でpsqlに接続

psql -U postgres

新しいユーザーを作成 (例: myuser)。パスワードは適宜設定。

CREATE USER myuser WITH PASSWORD ‘mypassword’;

新しいデータベースを作成 (例: mydatabase)。所有者を先ほど作成したユーザーに設定。

CREATE DATABASE mydatabase OWNER myuser;

psqlを終了

\q
“`

これで、ユーザーmyuserがデータベースmydatabaseに対する権限を持つようになります。Flaskアプリケーションからはこのユーザー情報を使って接続します。

基本的なデータベース接続（psycopg2を使用）

まず、ORMを使わずに、Pythonの低レベルなPostgreSQLアダプターであるpsycopg2を直接使ってデータベースに接続し、操作する方法を見てみましょう。これにより、データベース接続の基本的な流れと概念を理解できます。

psycopg2とは？

psycopg2は、Python Database API Specification (PEP 249) に準拠したPostgreSQLアダプターです。PythonコードからSQLクエリを実行し、結果を取得するための機能を提供します。C言語で実装されており、パフォーマンスが高いのが特徴です。

接続情報の定義

データベースに接続するには、以下の情報が必要です。

  • ホスト名 (host): データベースサーバーが動作しているマシンのIPアドレスまたはホスト名（例: localhost, 127.0.0.1, リモートサーバーのIP）
  • データベース名 (database): 接続したいデータベースの名前（例: mydatabase）
  • ユーザー名 (user): データベースに接続するためのユーザー名（例: myuser）
  • パスワード (password): そのユーザーのパスワード（例: mypassword）
  • ポート番号 (port): PostgreSQLがリッスンしているポート番号（デフォルトは5432）

これらの情報は、実際のアプリケーションでは設定ファイルや環境変数から読み込むべきですが、ここではコード内に直接記述します。

“`python

db_config.py (例)

DB_HOST = “localhost”
DB_NAME = “mydatabase”
DB_USER = “myuser”
DB_PASSWORD = “mypassword”
DB_PORT = “5432” # ポート番号は文字列でもintでも可
“`

psycopg2.connect() の使い方

psycopg2.connect() 関数を使ってデータベースに接続します。引数として上記の接続情報を文字列またはキーワード引数で渡します。

“`python
import psycopg2
from db_config import * # 先ほどの接続情報ファイルから読み込み

conn = None # 接続オブジェクト
cursor = None # カーソルオブジェクト

try:
# データベースに接続
conn = psycopg2.connect(
host=DB_HOST,
database=DB_NAME,
user=DB_USER,
password=DB_PASSWORD,
port=DB_PORT
)

print("データベースに接続しました！")

# カーソルを作成
# カーソルはSQLコマンドを実行し、結果を取得するために使われます。
cursor = conn.cursor()

# 簡単なクエリを実行して接続を確認
cursor.execute("SELECT version();")

# 結果を取得
db_version = cursor.fetchone()
print(f"PostgreSQLデータベースのバージョン: {db_version}")

except psycopg2.OperationalError as e:
print(f”データベース接続エラー: {e}”)
except Exception as e:
print(f”エラーが発生しました: {e}”)
finally:
# 接続とカーソルを閉じる (重要なクリーンアップ処理)
if cursor is not None:
cursor.close()
print(“カーソルを閉じました。”)
if conn is not None:
conn.close()
print(“データベース接続を閉じました。”)

print(“処理を終了します。”)
“`

psycopg2.connect() の引数:

  • dbname または database: データベース名
  • user: ユーザー名
  • password: パスワード
  • host: ホスト名またはIPアドレス
  • port: ポート番号
  • dsn: Data Source Name。接続情報を一つの文字列で指定することもできます（例: "dbname=mydatabase user=myuser password=mypassword host=localhost port=5432"）。

接続が成功したかどうかの確認:

psycopg2.connect() が例外を発生させずに完了すれば、接続は成功しています。例外が発生した場合は接続に失敗しています。特にpsycopg2.OperationalErrorは接続確立に関する問題（認証失敗、サーバーが見つからないなど）を示します。

カーソルオブジェクトについて

conn.cursor() で作成されるカーソルオブジェクトは、データベースとの実際のやり取り（SQLコマンドの送信、結果の受信）を行うためのオブジェクトです。SQLクエリはカーソルオブジェクトの execute() メソッドを使って実行します。

結果セットがあるクエリ（SELECT文など）の場合、カーソルオブジェクトのメソッドを使って結果を取得します。

  • fetchone(): 結果セットの次の1行を取得します。結果がない場合はNoneを返します。
  • fetchall(): 結果セットの全ての行をリストとして取得します。
  • fetchmany(size): 結果セットの次のsize行をリストとして取得します。

簡単なクエリの実行例

基本的なCRUD操作（作成、読み取り、更新、削除）をpsycopg2で行う例を見てみましょう。タスクを管理する簡単なテーブルを作成し、データを挿入・取得・更新・削除します。

“`python

tasks_psycopg2.py

import psycopg2
from db_config import *

def execute_query(query, params=None, fetch=False):
“””データベースクエリを実行する汎用関数”””
conn = None
cursor = None
result = None
try:
conn = psycopg2.connect(host=DB_HOST, database=DB_NAME, user=DB_USER, password=DB_PASSWORD, port=DB_PORT)
cursor = conn.cursor()

    # SQLインジェクションを防ぐためにパラメータは分離して渡す
    cursor.execute(query, params)

    if fetch:
        result = cursor.fetchall() # SELECTの場合は全件取得
    else:
        conn.commit() # INSERT, UPDATE, DELETEの場合はコミット

except psycopg2.OperationalError as e:
    print(f"データベース接続エラー: {e}")
except psycopg2.IntegrityError as e:
    print(f"整合性エラー: {e}")
    if conn:
        conn.rollback() # エラー時はロールバック
except Exception as e:
    print(f"エラーが発生しました: {e}")
    if conn:
        conn.rollback() # エラー時はロールバック
finally:
    if cursor is not None:
        cursor.close()
    if conn is not None:
        conn.close()

return result

1. テーブル作成

print(“— テーブル作成 —“)
create_table_query = “””
CREATE TABLE IF NOT EXISTS tasks (
id SERIAL PRIMARY KEY,
description VARCHAR(255) NOT NULL,
completed BOOLEAN DEFAULT FALSE
);
“””
execute_query(create_table_query)
print(“テーブル ‘tasks’ が存在することを確認または作成しました。”)

2. データ挿入 (Create)

print(“\n— データ挿入 —“)
insert_query = “INSERT INTO tasks (description) VALUES (%s);”
task1_desc = “牛乳を買う”
task2_desc = “Flaskチュートリアルを完了させる”
execute_query(insert_query, (task1_desc,)) # %s はプレースホルダ、パラメータはタプルで渡す
execute_query(insert_query, (task2_desc,))
print(“タスクを挿入しました。”)

3. データ取得 (Read)

print(“\n— データ取得 —“)
select_query = “SELECT id, description, completed FROM tasks;”
tasks = execute_query(select_query, fetch=True)
if tasks:
print(“全てのタスク:”)
for task in tasks:
print(f”ID: {task[0]}, 内容: {task[1]}, 完了: {task[2]}”)
else:
print(“タスクが見つかりませんでした。”)

IDが1のタスクを取得する例

select_one_query = “SELECT id, description, completed FROM tasks WHERE id = %s;”
task_id_to_fetch = 1
task = execute_query(select_one_query, (task_id_to_fetch,), fetch=True) # fetchmany/fetchoneでも良い
if task:
# fetchone()の場合はタプル、fetchall()の場合はタプルを要素とするリスト
print(f”\nID {task_id_to_fetch} のタスク: {task[0]}”) # fetchall()の場合のアクセス例
else:
print(f”\nID {task_id_to_fetch} のタスクは見つかりませんでした。”)

4. データ更新 (Update)

print(“\n— データ更新 —“)
update_query = “UPDATE tasks SET completed = %s WHERE id = %s;”
task_id_to_update = 1
execute_query(update_query, (True, task_id_to_update))
print(f”ID {task_id_to_update} のタスクを完了に更新しました。”)

更新後のタスクを確認

tasks_after_update = execute_query(select_query, fetch=True)
if tasks_after_update:
print(“更新後の全てのタスク:”)
for task in tasks_after_update:
print(f”ID: {task[0]}, 内容: {task[1]}, 完了: {task[2]}”)

5. データ削除 (Delete)

print(“\n— データ削除 —“)
delete_query = “DELETE FROM tasks WHERE id = %s;”
task_id_to_delete = 2
execute_query(delete_query, (task_id_to_delete,))
print(f”ID {task_id_to_delete} のタスクを削除しました。”)

削除後のタスクを確認

tasks_after_delete = execute_query(select_query, fetch=True)
if tasks_after_delete:
print(“削除後の全てのタスク:”)
for task in tasks_after_delete:
print(f”ID: {task[0]}, 内容: {task[1]}, 完了: {task[2]}”)
else:
print(“全てのタスクが削除されました。”)
“`

SQLインジェクション対策:

上の例で注目してほしいのは、SQLクエリに直接Pythonの文字列を埋め込むのではなく、execute(query, params)のようにクエリ文字列とパラメータを分離して渡している点です。psycopg2は、このパラメータを安全な方法でエスケープしてクエリに組み込みます。ユーザーからの入力を扱う際には、絶対に文字列フォーマットやf-stringなどで直接SQLクエリに埋め込まないでください。 必ずパラメータとして渡すようにしましょう。

トランザクション管理

データベース操作において、複数のSQLコマンドを一つの論理的な単位として扱うことがあります。これをトランザクションと呼びます。例えば、銀行口座AからBへ送金する場合、「Aから引き落とす」と「Bに入金する」という二つの操作が両方成功するか、あるいは両方失敗する（送金されなかったことにする）必要があります。片方だけが成功する状態は許されません。

psycopg2では、接続が確立された時点でトランザクションが開始されます。

  • conn.commit(): 現在のトランザクションで行われた全ての変更をデータベースに永続的に保存します。
  • conn.rollback(): 現在のトランザクションで行われた全ての変更を取り消し、トランザクション開始時点の状態に戻します。

INSERT, UPDATE, DELETEなどのデータを変更するクエリを実行した後は、必ずcommit()を呼び出して変更を確定させる必要があります。SELECT文のような読み取り専用のクエリでは通常コミットは不要です。（ただし、一部のデータベース設定では読み取り操作でもトランザクションを明示的に閉じる必要がある場合がありますが、一般的な使い方では変更クエリのみコミット/ロールバックが必要です）。

エラーが発生した場合は、データベースが一貫性のない状態になるのを防ぐためにrollback()を呼び出すのが一般的です。上記のexecute_query関数では、try...exceptブロックで例外をキャッチした場合にconn.rollback()を呼び出しています。

接続とカーソルのクローズ

データベースへの接続は有限のリソースです。使用後は必ず接続とカーソルを閉じる必要があります。cursor.close() および conn.close() を呼び出します。finallyブロック内でこれを行うことで、例外が発生した場合でも確実にクローズされるようにします。

また、Pythonのwithステートメントを使うと、リソース管理を自動で行ってくれます。psycopg2の接続とカーソルもwithに対応しています。

“`python
import psycopg2
from db_config import *

try:
# withを使うと、ブロックを抜けるときに自動でcloseされる
with psycopg2.connect(host=DB_HOST, database=DB_NAME, user=DB_USER, password=DB_PASSWORD, port=DB_PORT) as conn:
# with conn: のブロックを抜けるときに自動でcommitまたはrollbackされる
# デフォルトでは、エラーが発生しなければcommit、発生すればrollback

    with conn.cursor() as cursor:
        cursor.execute("SELECT 1;")
        result = cursor.fetchone()
        print(f"クエリ結果: {result}")

except psycopg2.OperationalError as e:
print(f”データベース接続エラー: {e}”)
except Exception as e:
print(f”エラーが発生しました: {e}”)

print(“withステートメントによる接続管理が終了しました。”)
“`

withステートメントはコードを簡潔にし、リソースリークを防ぐのに役立ちます。特に関数を抜ける際などにクローズし忘れる心配がなくなります。

Flaskアプリケーション内でのpsycopg2の利用

シンプルなFlaskアプリケーションでpsycopg2を使う場合、リクエストごとにデータベース接続を確立し、リクエスト処理が完了したら接続を閉じる、というパターンが考えられます。Flaskのgオブジェクトやbefore_request/after_requestデコレーターを利用できます。

“`python

app_psycopg2.py

from flask import Flask, g
import psycopg2
from db_config import *

app = Flask(name)
app.config.from_object(name) # app.config[‘DB_HOST’]などでアクセスできるようにする

環境変数などから読み込む場合はこちら

app.config[‘DB_HOST’] = os.environ.get(‘DB_HOST’, ‘localhost’)

… 同様に他の設定も読み込む

def get_db():
“””リクエストごとにデータベース接続を取得するヘルパー関数”””
if ‘db_conn’ not in g:
try:
g.db_conn = psycopg2.connect(
host=app.config[‘DB_HOST’],
database=app.config[‘DB_NAME’],
user=app.config[‘DB_USER’],
password=app.config[‘DB_PASSWORD’],
port=app.config[‘DB_PORT’]
)
# psycopg2のデフォルトではautocommit=False
# 明示的にcommit/rollbackを制御する必要がある
except psycopg2.OperationalError as e:
# 接続エラーを適切に処理する（例えば、HTTP 500エラーを返すなど）
print(f”データベース接続エラー: {e}”)
g.db_conn = None # エラー時は接続オブジェクトをNoneにする
# 必要に応じて例外を再 raise するか、エラーレスポンスを返す

return g.db_conn

@app.teardown_request
def close_db(error):
“””リクエストの終わりにデータベース接続を閉じる”””
db_conn = g.pop(‘db_conn’, None)
if db_conn is not None:
db_conn.close()
# print(“データベース接続を閉じました (teardown_request)。”) # デバッグ用

@app.route(‘/’)
def index():
db = get_db()
if db is None:
return “データベースに接続できませんでした。”, 500

cursor = None
try:
    cursor = db.cursor()
    cursor.execute("SELECT version();")
    db_version = cursor.fetchone()[0] # fetchoneはタプルを返すので最初の要素を取得
    return f"PostgreSQLデータベースに接続成功！バージョン: {db_version}"
except Exception as e:
    # クエリ実行エラーを処理
    print(f"クエリ実行エラー: {e}")
    # 例: データベースへの書き込みでエラーが発生した場合、ロールバックが必要
    # db.rollback() # 書き込み操作でエラーが発生した場合
    return f"クエリ実行中にエラーが発生しました: {e}", 500
finally:
    if cursor is not None:
        cursor.close()
        # SELECT文など読み取り専用の場合はcommit不要
        # INSERT/UPDATE/DELETEの場合はここでcommitするか、
        # エラーの場合はrollbackする処理が必要
        # 例： db.commit()
        # ただし、Flaskのリクエスト処理内で複雑なトランザクションを管理するのはpsycopg2直だと少し手間がかかる
        # ORM (SQLAlchemy) を使うとセッション管理が楽になる

“`

このアプローチは、リクエストごとに独立したデータベース接続を使用するため、単純なアプリケーションには適しています。しかし、トランザクション管理をリクエスト処理全体にわたって行う場合や、複数の関数で同じ接続オブジェクトを共有する必要がある場合は、gオブジェクトの利用とcommit/rollbackの適切な配置が重要になります。

psycopg2の欠点:

  • 生SQL: 全てのデータベース操作をSQLクエリとして記述する必要があります。これは柔軟性が高い反面、SQLの知識が必須であり、クエリの記述ミスが発生しやすいです。
  • オブジェクトマッピングがない: データベースの行（レコード）をPythonのオブジェクトとして自動的にマッピングする機能がありません。取得したデータはタプルやリストの形で扱われるため、データの属性に名前でアクセスするのが難しくなります（例: task[0]ではなくtask.idとしたい）。
  • データベースの種類に依存: データベースの種類を変更する場合（例: PostgreSQLからMySQLへ）、全てのSQLクエリを書き直す必要があります。
  • 定型コードが多い: 接続の確立、カーソルの作成、クエリ実行、結果取得、コミット/ロールバック、クローズといった一連の流れを毎回記述する必要があります。

これらの欠点を補い、より効率的でメンテナンス性の高いデータベースアクセスを実現するために、通常はORM (Object-Relational Mapper) を使用します。

ORMの導入：Flask-SQLAlchemyを使用

ORMは、オブジェクト指向プログラミング言語（Python）とリレーショナルデータベースの間に位置し、データベースのテーブルをクラスに、テーブルの行をオブジェクトにマッピングする技術です。これにより、Pythonオブジェクトを操作する感覚でデータベースを操作できるようになります。

ORMとは何か？なぜ使うのか？

ORM (Object-Relational Mapper):
オブジェクト指向プログラミング言語のオブジェクトと、リレーショナルデータベースのテーブル構造との間のマッピングを行うソフトウェア層です。

ORMを使うメリット:

  • 生産性の向上: SQLクエリを直接書く代わりに、オブジェクト操作でデータベースアクセスができます。これにより、開発速度が向上します。
  • SQLインジェクション対策: パラメータバインディングがORMによって抽象化されるため、適切に利用すればSQLインジェクションのリスクを低減できます。
  • データベース変更への対応: ORMは様々なデータベースシステムをサポートしています。データベースの種類を変更する際も、ORM層より上のアプリケーションコードの変更を最小限に抑えることができます（SQLの記述から解放されるため）。
  • 可読性とメンテナンス性: データベース構造がPythonのクラスとして表現されるため、コードが理解しやすくなり、メンテナンスが容易になります。

デメリット:

  • 学習コスト: ORM独自の概念（セッション、モデル定義、クエリ構文など）を習得する必要があります。
  • パフォーマンス: ORMが生成するSQLが常に最適であるとは限りません。複雑なクエリの場合、生のSQLを書いた方がパフォーマンスが優れる場合もあります。
  • 抽象化によるブラックボックス: ORMの内部動作（どのようなSQLが生成されるか）を理解していないと、パフォーマンス問題が発生した場合の原因特定や解決が難しくなることがあります。

SQLAlchemyとは？

SQLAlchemyは、Pythonで最も人気があり、機能豊富なORMライブラリです。様々なデータベースシステム（PostgreSQL, MySQL, SQLite, Oracleなど）をサポートしています。SQLAlchemyは強力な表現力を持ち、低レベルなSQL操作から高レベルなオブジェクトマッピングまで幅広く対応できます。

Flask-SQLAlchemyとは？

Flask-SQLAlchemyは、SQLAlchemyをFlaskアプリケーションに簡単に統合するための拡張機能です。以下の機能を提供します。

  • アプリケーション設定からのデータベースURIの読み込み
  • SQLAlchemyオブジェクトのFlaskアプリケーションへの組み込み
  • リクエストスコープでのセッション管理（リクエスト開始時にセッションを開始し、リクエスト終了時にコミットまたはロールバック）
  • ヘルパー関数やシグナルの提供

これにより、FlaskアプリケーションでSQLAlchemyを使う際の定型コードを減らし、よりFlaskらしい方法でデータベースを扱えるようになります。

インストール

既に「開発環境の準備」のセクションでpip install Flask-SQLAlchemyを実行済みのはずです。確認してみてください。

Flask-SQLAlchemyの設定方法

Flask-SQLAlchemyを使用するには、まずアプリケーションの設定でデータベース接続URIを指定する必要があります。

“`python

config.py

import os

class Config:
SECRET_KEY = os.environ.get(‘SECRET_KEY’, ‘a_very_secret_key_default’) # 秘密鍵は必須ではないが、多くのFlask拡張で推奨
SQLALCHEMY_DATABASE_URI = os.environ.get(‘DATABASE_URL’, ‘postgresql://myuser:mypassword@localhost:5432/mydatabase’)
SQLALCHEMY_TRACK_MODIFICATIONS = False # SQLAlchemyの変更追跡システムを無効にする (リソースを節約するため推奨)

環境変数から設定を読み込む場合は、以下のようにする

import os

class Config:

# SQLALCHEMY_DATABASE_URI = os.environ.get(‘DATABASE_URL’) or \

‘postgresql://myuser:mypassword@localhost:5432/mydatabase’

SQLALCHEMY_TRACK_MODIFICATIONS = False

SECRET_KEY = os.environ.get(‘SECRET_KEY’, ‘default_secret_key’) # 環境変数からの読み込み推奨

環境変数 DATABASE_URL の例: postgresql://myuser:mypassword@localhost:5432/mydatabase

“`

SQLALCHEMY_DATABASE_URI:

この設定は、データベースの種類、接続ユーザー、パスワード、ホスト、ポート、データベース名を以下のフォーマットで指定します。

dialect+driver://user:password@host:port/database

  • dialect: データベースの種類 (postgresql, mysql, sqlite, etc.)
  • driver (オプション): 使用するDBアダプター (psycopg2, mysqlconnector, etc.)。postgresqlの場合、デフォルトはpsycopg2なので省略可能です (postgresql://...でOK)。
  • user, password, host, port, database: 接続情報

例: postgresql://myuser:mypassword@localhost:5432/mydatabase

SQLALCHEMY_TRACK_MODIFICATIONS:

この設定は、SQLAlchemyの変更追跡システムを有効にするかどうかを指定します。ほとんどの場合、このシステムは不要であり、不要なシグナルを発行してリソースを消費する可能性があるため、Falseに設定することを強く推奨します。

SQLAlchemy オブジェクトの初期化

Flaskアプリケーションオブジェクトと設定情報をSQLAlchemyオブジェクトに渡して初期化します。

“`python

app.py

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from config import Config # 先ほど作成した設定ファイル

app = Flask(name)
app.config.from_object(Config) # Configクラスから設定を読み込む

db = SQLAlchemy(app) # FlaskアプリケーションにSQLAlchemyを連携させる
“`

このdbオブジェクトが、モデル定義やデータベース操作の中心となります。

モデルの定義

SQLAlchemyでは、データベースのテーブルをPythonのクラスとして定義します。このクラスを「モデル」と呼びます。モデルはdb.Modelを継承し、テーブルのカラムはdb.Columnとして定義します。

“`python

models.py

from app import db # app.pyで初期化したdbオブジェクトをインポート

class Task(db.Model):
tablename = ‘tasks’ # テーブル名を指定 (指定しない場合はクラス名が小文字になる)

id = db.Column(db.Integer, primary_key=True) # 主キー、SERIAL型にマッピングされる
description = db.Column(db.String(255), nullable=False) # VARCHAR(255)、NULLを許可しない
completed = db.Column(db.Boolean, default=False, nullable=False) # BOOLEAN、デフォルト値はFalse、NULLを許可しない

# オブジェクトの文字列表現を定義（オプションだが便利）
def __repr__(self):
    return f'<Task {self.id}: {self.description}>'

他のモデルも同様に定義する

class User(db.Model):

id = db.Column(db.Integer, primary_key=True)

username = db.Column(db.String(80), unique=True, nullable=False)

email = db.Column(db.String(120), unique=True, nullable=False)

def repr(self):

return f’‘

“`

db.Columnの主な引数:

  • 第一引数: カラムのデータ型 (db.Integer, db.String, db.Boolean, db.DateTime, db.Float, db.Textなど)
  • primary_key=True: そのカラムを主キーとして設定します。
  • nullable=False: そのカラムにNULL値を許可しません。デフォルトはTrueです。
  • unique=True: そのカラムの値が一意である必要があります。
  • default=value: カラムのデフォルト値を設定します。
  • server_default=value: データベースレベルでのデフォルト値を設定します（例: server_default=db.func.now()で現在時刻）。
  • index=True: そのカラムにインデックスを作成します（検索性能向上に役立ちます）。
  • ForeignKey('tablename.columnname'): 外部キーを設定します。

データベースの作成 (db.create_all())

モデルを定義した後、これらのモデル定義に基づいてデータベース内にテーブルを作成できます。

“`python

create_db.py (一度だけ実行するスクリプト)

from app import app, db # Flaskアプリケーションとdbオブジェクトをインポート
from models import * # 定義したモデルをインポート

with app.app_context(): # Flaskアプリケーションコンテキスト内で実行する必要がある
db.create_all() # 全てのdb.Modelを継承したクラスに基づいてテーブルを作成
print(“データベーステーブルを作成しました。”)
“`

このスクリプトを仮想環境内で実行すると、config.pyで指定したデータベースにtasksテーブルが作成されます。

bash
(venv) python create_db.py

注意: db.create_all()は、テーブルが既に存在する場合は何もしません。テーブル構造の変更（カラムの追加・削除・変更など）には対応していません。テーブル構造の変更を管理するには、Alembicのようなマイグレーションツールを使用するのが一般的です。

データの操作 (CRUD)

定義したモデルを使って、データベースのデータを操作します。全てのデータベース操作は「セッション」を通じて行われます。

セッション管理 (db.session):

セッションは、データベースとの対話（クエリの送信、変更の追跡、コミット、ロールバック）を行うための中心的なオブジェクトです。SQLAlchemyは、変更をすぐにデータベースに書き込むのではなく、まずセッションに追加（add()）、変更、削除（delete()）します。これらの変更は、db.session.commit()が呼び出されたときにまとめてデータベースに書き込まれます。db.session.rollback()を呼び出すと、現在のセッションで行われた未コミットの変更は全て破棄されます。

Flask-SQLAlchemyは、リクエストスコープでセッションを自動的に管理します。つまり、各Webリクエストの開始時に新しいセッションが作成され、リクエストの終了時（レスポンスを返す直前）にセッションが自動的にコミットされるか、例外が発生した場合はロールバックされます。これにより、Webアプリケーションでのセッション管理が非常に簡単になります。開発者は通常、明示的にdb.session.close()を呼び出す必要はありません。

1. データ挿入 (Create):

モデルのインスタンスを作成し、db.session.add()でセッションに追加し、db.session.commit()でデータベースに書き込みます。

“`python

app.py (routesの例)

from flask import render_template, request, redirect, url_for
from app import app, db # app.pyで初期化したdbオブジェクトをインポート
from models import Task # 定義したモデルをインポート

@app.route(‘/tasks/new’, methods=[‘GET’, ‘POST’])
def new_task():
if request.method == ‘POST’:
description = request.form.get(‘description’)
if description:
# Taskモデルのインスタンスを作成
new_task = Task(description=description)

        # セッションに追加
        db.session.add(new_task)

        # データベースにコミット
        try:
            db.session.commit()
            print("新しいタスクが追加されました！")
            return redirect(url_for('list_tasks')) # タスク一覧ページにリダイレクト
        except Exception as e:
            db.session.rollback() # エラー時はロールバック
            print(f"タスクの追加中にエラーが発生しました: {e}")
            # エラーメッセージを表示するなど、適切なエラー処理を行う
            return "タスクの追加に失敗しました。", 500
    else:
        return "タスクの内容を入力してください。", 400

# GETリクエストの場合はタスク追加フォームを表示
return render_template('new_task.html')

@app.route(‘/tasks’)
def list_tasks():
# データベースから全てのタスクを取得
tasks = Task.query.all()
return render_template(‘list_tasks.html’, tasks=tasks)
“`

2. データ取得 (Read):

モデルクラスの.query属性を使って、データベースからデータを取得します。.queryはSQLAlchemyのQueryオブジェクトを返します。

“`python

models.py と app.py は上記のものを使用

@app.route(‘/tasks’)
def list_tasks():
# Task.query は Task モデルに対するクエリビルダ
# .all() はクエリ結果の全てのオブジェクトをリストとして取得
all_tasks = Task.query.all()

# IDで取得 (.get() は主キーでの取得に最適化されている)
task_by_id = Task.query.get(1) # id=1 のタスクを取得

# 条件を指定して取得 (.filter_by() はキーワード引数で単純な等価条件を指定)
completed_tasks = Task.query.filter_by(completed=True).all()
# WHERE completed = TRUE に相当

# 条件を指定して取得 (.filter() はSQL式を柔軟に指定)
# desc LIKE '%チュートリアル%' なタスクを取得
tutorial_tasks = Task.query.filter(Task.description.like('%チュートリアル%')).all()
# WHERE description LIKE '%チュートリアル%' に相当

# 複数の条件を指定
incomplete_urgent_tasks = Task.query.filter_by(completed=False, urgent=True).all() # urgentカラムがあるとして

# OR条件を指定 (SQLAlchemyの or_ を使う)
from sqlalchemy import or_
urgent_or_incomplete_tasks = Task.query.filter(or_(Task.urgent == True, Task.completed == False)).all()

# 結果の並べ替え (.order_by())
tasks_ordered_by_id = Task.query.order_by(Task.id).all() # 昇順
tasks_ordered_by_id_desc = Task.query.order_by(Task.id.desc()).all() # 降順

# 結果の件数を制限 (.limit(), .offset())
first_three_tasks = Task.query.limit(3).all()
tasks_after_fifth = Task.query.offset(5).all()
tasks_from_6_to_10 = Task.query.offset(5).limit(5).all() # 6件目から5件

# 最初の1件を取得 (.first())
first_incomplete_task = Task.query.filter_by(completed=False).first() # WHERE completed = FALSE LIMIT 1

# 件数をカウント (.count()) - 非推奨、.scalar(db.select(db.func.count()).select_from(Task)) を推奨
# Flask-SQLAlchemyでは Task.query.count() がまだ使えますが、将来的に非推奨になる可能性があります
# 可能な限り SQLAlchemy 2.0 スタイルの Select を使うことが推奨されますが、
# Flask-SQLAlchemyではTask.queryによる既存のコードも広く使われています。
# ここでは .query スタイルを中心に説明します。
total_tasks_count = Task.query.count()

# select文を明示的に使う場合 (SQLAlchemy 2.0 style)
# from sqlalchemy import select
# stmt = select(Task).where(Task.completed == True)
# completed_tasks_v2 = db.session.execute(stmt).scalars().all()
# stmt_count = select(db.func.count()).select_from(Task)
# total_tasks_count_v2 = db.session.execute(stmt_count).scalar()


# ここでは、単純に全件取得してテンプレートに渡す例を維持
tasks = Task.query.all()
return render_template('list_tasks.html', tasks=tasks)

“`

様々なクエリメソッド:

メソッド 説明 例
all() クエリ結果の全ての行をオブジェクトのリストとして取得します。 Task.query.all()
first() クエリ結果の最初の1行をオブジェクトとして取得します。結果がない場合はNoneを返します。LIMIT 1に相当。 Task.query.filter_by(completed=False).first()
get(id) 主キーの値に基づいてオブジェクトを1つ取得します。見つからない場合はNoneを返します。 Task.query.get(5)
filter_by(**kwargs) キーワード引数で単純な等価条件を指定します。 Task.query.filter_by(completed=True, user_id=1)
filter(*criterion) より複雑な条件をSQL式で指定します。 Task.query.filter(Task.description.like('% urgent%'))
order_by(*criterion) 結果の並べ替えを指定します。カラム名またはカラム名.desc()を使います。 Task.query.order_by(Task.id.desc())
limit(limit) 結果の最大件数を制限します。 Task.query.limit(10)
offset(offset) 結果の先頭から指定した件数だけスキップします。ページネーションなどに使用します。 Task.query.offset(20)
count() クエリ結果の行数をカウントします。（非推奨、SQLAlchemy 2.0スタイルを推奨） Task.query.count()
delete(synchronize_session='evaluate') クエリに一致する全ての行をまとめて削除します。セッションと同期するメカニズムを指定できます。 Task.query.filter_by(completed=True).delete()
update(values, synchronize_session='evaluate') クエリに一致する全ての行をまとめて更新します。値を辞書で指定します。セッションと同期するメカニズムを指定できます。 Task.query.filter_by(completed=True).update({'completed': False})

filter()メソッドは非常に強力で、sqlalchemy.sql.and_, sqlalchemy.sql.or_, sqlalchemy.sql.not_ などの関数を使って複雑なAND/OR条件を構築できます。

3. データ更新 (Update):

更新したいオブジェクトを取得し、その属性を変更します。変更はセッションによって追跡されます。db.session.commit()を呼び出すことで、変更がデータベースに書き込まれます。

“`python

app.py (routesの例)

@app.route(‘/tasks//complete’)
def complete_task(task_id):
# IDでタスクを取得
task = Task.query.get(task_id)

if task:
    # オブジェクトの属性を変更
    task.completed = True

    # セッションに変更が追跡されているので、commitするだけでOK
    try:
        db.session.commit()
        print(f"タスク ID {task_id} を完了にしました。")
        return redirect(url_for('list_tasks'))
    except Exception as e:
        db.session.rollback() # エラー時はロールバック
        print(f"タスクの更新中にエラーが発生しました: {e}")
        return "タスクの更新に失敗しました。", 500
else:
    return "タスクが見つかりませんでした。", 404

例: 編集フォームを使った更新

@app.route(‘/tasks//edit’, methods=[‘GET’, ‘POST’])
def edit_task(task_id):
task = Task.query.get_or_404(task_id) # get_or_404 は見つからない場合に404エラーを返す

if request.method == 'POST':
    description = request.form.get('description')
    completed = request.form.get('completed') == 'on' # チェックボックスの値は 'on' または None

    if description:
        task.description = description
        task.completed = completed

        try:
            db.session.commit()
            print(f"タスク ID {task_id} が更新されました。")
            return redirect(url_for('list_tasks'))
        except Exception as e:
            db.session.rollback()
            print(f"タスクの更新中にエラーが発生しました: {e}")
            return "タスクの更新に失敗しました。", 500
    else:
        return "タスクの内容を入力してください。", 400

# GETリクエストの場合は編集フォームを表示
return render_template('edit_task.html', task=task)

“`

4. データ削除 (Delete):

削除したいオブジェクトを取得し、db.session.delete()でセッションから削除対象としてマークし、db.session.commit()でデータベースから削除します。

“`python

app.py (routesの例)

@app.route(‘/tasks//delete’, methods=[‘POST’]) # 通常はPOSTやDELETEメソッドで削除
def delete_task(task_id):
task = Task.query.get(task_id)

if task:
    # セッションから削除対象としてマーク
    db.session.delete(task)

    # データベースから削除を実行
    try:
        db.session.commit()
        print(f"タスク ID {task_id} が削除されました。")
        return redirect(url_for('list_tasks'))
    except Exception as e:
        db.session.rollback() # エラー時はロールバック
        print(f"タスクの削除中にエラーが発生しました: {e}")
        return "タスクの削除に失敗しました。", 500
else:
    return "タスクが見つかりませんでした。", 404

“`

セッション管理の重要性と Flask-SQLAlchemy の役割

SQLAlchemyのセッションは、データベース操作における作業単位（Unit of Work）の概念を表します。db.session.add(), db.session.delete(), モデル属性の変更などは、セッション内で「保留中の変更」として追跡されます。これらの変更は、db.session.commit()が呼び出されるまでデータベースには反映されません。

commit()は、保留中の全ての変更をアトミックに（全て成功するか、全て失敗するか）データベースに書き込みます。これにより、データベースが一貫性のない状態になるのを防ぎます。

rollback()は、現在のセッションで行われた全ての未コミットの変更を取り消します。データベースエラーなどが発生した場合に、不正なデータがデータベースに書き込まれるのを防ぐために使用されます。

Flask-SQLAlchemyは、Webアプリケーションのリクエスト・レスポンスサイクルに合わせて、このセッション管理を自動で行ってくれます。

  1. リクエスト開始前: before_requestハンドラで新しいセッション（db.session）が作成されます。
  2. リクエスト処理中: アプリケーションコードはdb.sessionを使ってデータベース操作を行います。
  3. リクエスト正常終了時: after_requestハンドラでdb.session.commit()が自動的に呼び出されます。
  4. リクエスト処理中に例外発生時: teardown_requestハンドラでdb.session.rollback()が自動的に呼び出されます。
  5. リクエスト終了後: teardown_requestハンドラでセッションがクローズされます（db.session.close()）。

この自動化されたセッション管理により、開発者はほとんどの場合、db.session.commit()やdb.session.rollback()、db.session.close()を明示的に書く必要がありません。例外処理などで明示的にrollback()を呼び出す必要がある場面はありますが、基本的にはFlask-SQLAlchemyに任せることができます。これはFlaskアプリケーションでSQLAlchemyを使う大きなメリットの一つです。

実践的なアプリケーション開発（簡単なタスク管理アプリ）

これまでに学んだことを使って、簡単なタスク管理Webアプリケーションを構築してみましょう。プロジェクト構造を整理し、FlaskのBlueprintも導入します。

プロジェクト構造

flask-postgres-tutorial/
├── venv/ # 仮想環境
├── config.py # アプリケーション設定
├── create_db.py # データベーステーブル作成用スクリプト (一度実行)
├── app.py # FlaskアプリケーションインスタンスとSQLAlchemyオブジェクトの初期化
├── models.py # データベースモデルの定義
├── tasks/ # タスク関連のBlueprintディレクトリ
│ ├── __init__.py # Blueprint初期化
│ └── routes.py # ルート定義 (ビュー関数)
├── templates/ # HTMLテンプレート
│ ├── base.html # 基本レイアウト
│ ├── list_tasks.html # タスク一覧表示
│ ├── new_task.html # タスク追加フォーム
│ └── edit_task.html # タスク編集フォーム
└── run.py # アプリケーション実行用スクリプト

config.py

“`python

config.py

import os

file は現在のファイル (config.py) を指す

os.path.abspath(file) は config.py の絶対パス

os.path.dirname(…) はそのディレクトリ (flask-postgres-tutorial) を指す

basedir = os.path.abspath(os.path.dirname(file))

class Config:
# 秘密鍵は本番環境ではランダムな文字列を環境変数から設定すべき
SECRET_KEY = os.environ.get(‘SECRET_KEY’) or ‘you-will-never-guess’

# PostgreSQL接続URI。環境変数 DATABASE_URL が設定されていなければローカルDBに接続。
# 例: export DATABASE_URL="postgresql://myuser:mypassword@localhost:5432/mydatabase"
SQLALCHEMY_DATABASE_URI = os.environ.get('DATABASE_URL') or \
    'postgresql://myuser:mypassword@localhost:5432/mydatabase'

SQLALCHEMY_TRACK_MODIFICATIONS = False # 変更追跡システムを無効化
SQLALCHEMY_ECHO = True # 生成されるSQLをコンソールに出力 (デバッグ用、本番ではFalse推奨)

“`

app.py

“`python

app.py

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from config import Config

app = Flask(name)
app.config.from_object(Config) # 設定を読み込む

db = SQLAlchemy(app) # dbオブジェクトを初期化

Blueprintの登録はappオブジェクトの作成後に行う

from tasks.routes import tasks_bp # tasks Blueprintをインポート
app.register_blueprint(tasks_bp, url_prefix=’/tasks’) # /tasks というURLプレフィックスで登録

アプリケーションインスタンスとdbオブジェクトを他のモジュールからインポートできるようにしておく

“`

models.py

“`python

models.py

from app import db # app.pyで初期化したdbオブジェクトをインポート

class Task(db.Model):
tablename = ‘tasks’ # テーブル名を指定

id = db.Column(db.Integer, primary_key=True)
description = db.Column(db.String(255), nullable=False)
completed = db.Column(db.Boolean, default=False, nullable=False)
created_at = db.Column(db.DateTime, server_default=db.func.now()) # 作成日時を自動記録

def __repr__(self):
    return f'<Task {self.id}: {self.description} (完了: {self.completed})>'

``created_atカラムを追加しました。server_default=db.func.now()`は、行が挿入されたときにデータベース側で現在時刻を自動で設定するように指示します。

tasks/__init__.py

Blueprintを初期化するファイルです。

“`python

tasks/init.py

from flask import Blueprint

‘tasks’という名前でBlueprintを作成し、このモジュールの名前空間を指定

tasks_bp = Blueprint(‘tasks’, name, template_folder=’../templates’)

routesファイルをインポートすることで、ルートがBlueprintに登録される

from . import routes
“`

tasks/routes.py

タスクに関連するビュー関数を定義します。Blueprintオブジェクト（tasks_bp）を使います。

“`python

tasks/routes.py

from flask import render_template, request, redirect, url_for
from . import tasks_bp # init.pyで定義したBlueprintをインポート
from app import db # app.pyで初期化したdbオブジェクトをインポート
from models import Task # 定義したモデルをインポート

@tasks_bp.route(‘/’)
def list_tasks():
# 全てのタスクを作成日時が新しい順に取得
tasks = Task.query.order_by(Task.created_at.desc()).all()
return render_template(‘list_tasks.html’, tasks=tasks)

@tasks_bp.route(‘/new’, methods=[‘GET’, ‘POST’])
def new_task():
if request.method == ‘POST’:
description = request.form.get(‘description’)
if description:
new_task = Task(description=description)
db.session.add(new_task)
try:
db.session.commit()
# Blueprint内で url_for を使う場合、Endpoint名にBlueprint名をプレフィックスとして付ける
# (Blueprint名).(ルート関数名) => ‘tasks.list_tasks’
return redirect(url_for(‘tasks.list_tasks’))
except Exception as e:
db.session.rollback()
print(f”タスクの追加中にエラーが発生しました: {e}”)
# TODO: エラーメッセージをユーザーに表示する仕組み (Flask-Flashなど)
return “タスクの追加に失敗しました。”, 500
else:
# TODO: エラーメッセージをユーザーに表示する仕組み
return render_template(‘new_task.html’, error=”タスクの内容を入力してください。”), 400

return render_template('new_task.html')

@tasks_bp.route(‘//edit’, methods=[‘GET’, ‘POST’])
def edit_task(task_id):
# get_or_404() はオブジェクトが見つからなければ自動で404レスポンスを返す便利なメソッド
task = Task.query.get_or_404(task_id)

if request.method == 'POST':
    description = request.form.get('description')
    completed = request.form.get('completed') == 'on' # チェックボックスの値

    if description:
        task.description = description
        task.completed = completed

        try:
            db.session.commit()
            print(f"タスク ID {task_id} が更新されました。")
            return redirect(url_for('tasks.list_tasks'))
        except Exception as e:
            db.session.rollback()
            print(f"タスクの更新中にエラーが発生しました: {e}")
             # TODO: エラーメッセージ
            return "タスクの更新に失敗しました。", 500
    else:
        # TODO: エラーメッセージ
        return render_template('edit_task.html', task=task, error="タスクの内容を入力してください。"), 400


return render_template('edit_task.html', task=task)

@tasks_bp.route(‘//delete’, methods=[‘POST’])
def delete_task(task_id):
task = Task.query.get_or_404(task_id)

db.session.delete(task)
try:
    db.session.commit()
    print(f"タスク ID {task_id} が削除されました。")
    return redirect(url_for('tasks.list_tasks'))
except Exception as e:
    db.session.rollback()
    print(f"タスクの削除中にエラーが発生しました: {e}")
    # TODO: エラーメッセージ
    return "タスクの削除に失敗しました。", 500

“`

テンプレート (templates/)

簡単なHTMLテンプレートを作成します。Jinja2テンプレートエンジンを使用します。

templates/base.html (基本レイアウト)

“`html






{% block title %}Flask タスク管理{% endblock %}


{% block content %}{% endblock %}

“`

templates/list_tasks.html (タスク一覧)

“`html
{% extends ‘base.html’ %}

{% block title %}タスク一覧 – Flask タスク管理{% endblock %}

{% block content %}

タスク一覧

{% if tasks %}
    {% for task in tasks %}
        <div class="task-item">
            <span class="{{ 'completed' if task.completed }}">
                {{ task.description }} (作成: {{ task.created_at.strftime('%Y-%m-%d %H:%M') if task.created_at else '日時不明' }})
            </span>

            {% if not task.completed %}
                <a href="{{ url_for('tasks.edit_task', task_id=task.id) }}">編集</a>
                <form action="{{ url_for('tasks.delete_task', task_id=task.id) }}" method="post" onsubmit="return confirm('このタスクを削除してもよろしいですか？');">
                    <button type="submit">削除</button>
                </form>
            {% else %}
                <span style="color: green;">完了</span>
                <form action="{{ url_for('tasks.delete_task', task_id=task.id) }}" method="post" onsubmit="return confirm('このタスクを削除してもよろしいですか？');">
                    <button type="submit">削除</button>
                </form>
            {% endif %}
        </div>
    {% endfor %}
{% else %}
    <p>タスクはまだありません。</p>
{% endif %}

{% endblock %}
“`

templates/new_task.html (タスク追加フォーム)

“`html
{% extends ‘base.html’ %}

{% block title %}新しいタスク – Flask タスク管理{% endblock %}

{% block content %}

新しいタスクを追加

{% if error %}
    <p style="color: red;">{{ error }}</p>
{% endif %}

<form method="post">
    <label for="description">タスク内容:</label><br>
    <input type="text" id="description" name="description" size="50" required><br><br>

    <button type="submit">追加</button>
</form>

{% endblock %}
“`

templates/edit_task.html (タスク編集フォーム)

“`html
{% extends ‘base.html’ %}

{% block title %}タスク編集 – Flask タスク管理{% endblock %}

{% block content %}

タスク編集 (ID: {{ task.id }})

 {% if error %}
    <p style="color: red;">{{ error }}</p>
{% endif %}

<form method="post">
    <label for="description">タスク内容:</label><br>
    <input type="text" id="description" name="description" value="{{ task.description }}" size="50" required><br><br>

    <input type="checkbox" id="completed" name="completed" {% if task.completed %}checked{% endif %}>
    <label for="completed">完了</label><br><br>

    <button type="submit">更新</button>
</form>

{% endblock %}
“`

run.py

アプリケーションを実行するためのエントリポイントです。

“`python

run.py

from app import app

if name == ‘main‘:
# デバッグモードを有効にして開発サーバーを実行
app.run(debug=True)
“`

アプリケーションの実行手順

  1. 仮想環境を有効化: source venv/bin/activate (またはWindowsの対応コマンド)
  2. 必要なライブラリをインストール: pip install -r requirements.txt (requirements.txtを作成していない場合は pip install Flask psycopg2-binary Flask-SQLAlchemy)
  3. 環境変数を設定 (オプションだが推奨):
    • Linux/macOS: export DATABASE_URL="postgresql://myuser:mypassword@localhost:5432/mydatabase"
    • Windows (cmd): set DATABASE_URL=postgresql://myuser:mypassword@localhost:5432/mydatabase
    • Windows (PowerShell): $env:DATABASE_URL="postgresql://myuser:mypassword@localhost:5432/mydatabase"
    • 環境変数を使わない場合は、config.pyのデフォルト設定が使われます。
  4. データベーステーブルを作成: python create_db.py
  5. アプリケーションを実行: python run.py

サーバーが起動したら、ブラウザで http://127.0.0.1:5000/tasks/ にアクセスしてタスク一覧画面を確認できます。「新しいタスク」からタスクを追加したり、編集・削除を試したりしてください。

SQLAlchemyを使った各操作の実装コード詳細の振り返り

上記の例では、Flask-SQLAlchemyを使ってデータベース操作がどのように行われているかを示しました。

  • 追加: Task(description=...)でモデルインスタンスを作成し、db.session.add(instance)でセッションに追加、最後にdb.session.commit()でデータベースに書き込み。
  • 取得: Task.queryオブジェクトのメソッド（.all(), .get(), .filter_by(), .order_by(), etc.）を使ってデータを取得。結果はモデルインスタンスのリストまたは単一のインスタンスとして取得される。
  • 更新: Task.queryで更新したいモデルインスタンスを取得し、インスタンスの属性値を直接変更。最後にdb.session.commit()でデータベースに変更を反映。
  • 削除: Task.queryで削除したいモデルインスタンスを取得し、db.session.delete(instance)でセッションから削除対象としてマーク、最後にdb.session.commit()でデータベースから削除。

全ての書き込み操作（追加、更新、削除）の後には、必ずdb.session.commit()を呼び出す必要があることに注意してください。読み取り操作（取得）だけの場合はコミットは不要です。エラーが発生した場合は、db.session.rollback()を呼び出して未コミットの変更を取り消すことが重要です。Flask-SQLAlchemyはリクエスト終了時の自動コミット/ロールバックを提供しますが、複雑なロジック内やエラー処理で明示的に呼び出すこともあります。

エラーハンドリングとデバッグ

データベース関連のエラーはアプリケーションの安定性に大きく影響します。適切なエラーハンドリングとデバッグ方法を知っておくことは重要です。

データベース接続エラー

psycopg2.connect()やdb.init_app()（内部的に接続を試みる場合）で最も発生しやすいエラーは、psycopg2.OperationalErrorです。これは、データベースサーバーが見つからない、認証情報が間違っている、ネットワークに問題があるといった接続レベルの問題を示します。

Flaskアプリケーションの起動時や最初のリクエストで接続エラーが発生した場合、ユーザーには適切なエラーメッセージ（例: 500 Internal Server Error）を表示する必要があります。アプリケーション起動前に接続を確認する、あるいはリクエストごとの接続試行でエラーを捕捉し、 gracefully なエラーページにリダイレクトするといった対応が考えられます。

Flask-SQLAlchemyを使用している場合、app.config['SQLALCHEMY_DATABASE_URI']の設定が間違っていると、アプリケーション起動時にエラーが発生するか、最初のデータベースアクセス時にエラーが発生します。

SQLAlchemy固有のエラー

SQLAlchemyやデータベースレベルで発生する可能性のある主なエラータイプ：

  • sqlalchemy.exc.IntegrityError: データベースの制約違反（例: NOT NULL制約に反するNULL値の挿入、UNIQUE制約に反する重複データの挿入、外部キー制約違反など）が発生した場合。
  • sqlalchemy.exc.DataError: データ型の不一致など、データの形式に関する問題。
  • sqlalchemy.exc.ProgrammingError: 無効なSQL文の実行、存在しないテーブル/カラムへのアクセスなど、プログラミング上の誤り。
  • sqlalchemy.exc.OperationalError: データベースサーバー側の問題（例: 接続切れ、トランザクションのデッドロック、ディスクフルなど）。

これらのエラーが発生した場合、特にIntegrityErrorなどの書き込み操作に関するエラーでは、現在のセッションは不正な状態になる可能性があるため、必ずdb.session.rollback()を呼び出す必要があります。ロールバックせずにコミットしようとすると、別のエラーが発生する可能性があります。

上記のタスク管理アプリの例では、new_task, edit_task, delete_task ルートでtry...except Exceptionブロック内でdb.session.rollback()を呼び出しています。これは一般的なプラクティスです。より詳細なエラーハンドリングでは、エラーの種類（例: IntegrityError）によって異なる処理を行うこともあります。

“`python

例: IntegrityError のハンドリング

from sqlalchemy.exc import IntegrityError

try:
db.session.add(new_task)
db.session.commit()
except IntegrityError:
db.session.rollback() # ロールバックが必要
# 例: UNIQUE制約違反の場合、ユーザー名が重複しているなどのエラーメッセージを表示
print(“エラー: 重複したデータです。”)
# 適切なエラーレスポンスやメッセージを返す
except Exception as e:
db.session.rollback() # その他のエラーでもロールバック
print(f”予期しないエラー: {e}”)
# 適切なエラーレスポンスやメッセージを返す
“`

Flaskのデバッグモードと SQLAlchemy のクエリログ

  • Flaskのデバッグモード (app.run(debug=True) または app.config['DEBUG'] = True): デバッグモードを有効にすると、エラー発生時にブラウザにトレースバックを含む詳細なデバッグ情報が表示されます。開発中は非常に役立ちますが、本番環境では必ず無効にしてください。
  • SQLAlchemyのクエリログ (app.config['SQLALCHEMY_ECHO'] = True): この設定をTrueにすると、SQLAlchemyが実行する全てのSQLクエリとそのパラメータがコンソールに出力されます。これにより、ORMがどのようなSQLを生成しているかを確認でき、パフォーマンス問題やクエリの誤りをデバッグするのに役立ちます。これも本番環境では通常無効にします。

パフォーマンスとスケーリングの考慮事項

データベースのパフォーマンスはWebアプリケーションのレスポンス時間に直結します。規模が大きくなるにつれて、以下の点を考慮する必要があります。

N+1問題

これはORMを使う際によく発生するパフォーマンス問題です。例えば、「全てのユーザーとそのユーザーが作成した最新のタスクを取得したい」という場合に、まず全てのユーザーを1回のクエリで取得し、その後取得したユーザー一人ひとりに対して最新タスクを取得するクエリを個別に発行してしまうパターンです。

“`python

N+1問題の例（UserモデルとTaskモデルにリレーションがあるとして）

users = User.query.all() # 1回目のクエリ (全てのユーザーを取得)
for user in users:
# 各ユーザーに対してタスクを取得 (N回目のクエリ)
latest_task = user.tasks.order_by(Task.created_at.desc()).first()
print(f”{user.username}: {latest_task.description if latest_task else ‘タスクなし’}”)
“`

ユーザーがN人いる場合、合計で1+N回のクエリが発生します。Nが大きくなると、データベースへの負荷が高まり、パフォーマンスが著しく低下します。

対策:

SQLAlchemyでは、Eager Loading という手法を使ってこの問題を回避できます。親オブジェクトを取得する際に、関連する子オブジェクトもまとめて取得するようにORMに指示します。

  • joinedload(): JOINを使って関連データを一度に取得します。
    python
    # joinedload を使った例 (Userモデルに 'tasks' リレーションが定義されているとして)
    from sqlalchemy.orm import joinedload
    users_with_tasks = User.query.options(joinedload(User.tasks)).all()
    # 1回のクエリでユーザーと関連するタスクをJOINして取得
    for user in users_with_tasks:
    # tasks 属性にアクセスしても追加のクエリは発生しない
    latest_task = sorted(user.tasks, key=lambda t: t.created_at, reverse=True)[0] if user.tasks else None
    print(f"{user.username}: {latest_task.description if latest_task else 'タスクなし'}")
  • subqueryload(): 別のサブクエリを使って関連データを取得します。複雑なクエリや制限付きリレーションの場合に有用です。

インデックスの利用

WHERE句やORDER BY句で頻繁に使用されるカラムには、データベースインデックスを作成することで検索やソートの速度を大幅に向上させることができます。

SQLAlchemyモデルでインデックスを作成するには、db.Columnの定義時にindex=Trueを指定するか、__table_args__で複合インデックスなどを定義します。

“`python
class Task(db.Model):
# … 他のカラム …
description = db.Column(db.String(255), nullable=False, index=True) # descriptionにインデックス
completed = db.Column(db.Boolean, default=False, nullable=False, index=True) # completedにインデックス

# 複合インデックスの例: completedとcreated_atの組み合わせでソートやフィルタリングが多い場合
__table_args__ = (
    db.Index('idx_completed_created_at', 'completed', 'created_at'),
)

“`

インデックスは検索を速くしますが、データの書き込み（INSERT, UPDATE, DELETE）にはオーバーヘッドが発生します。適切なカラムにのみインデックスを作成することが重要です。

データベース接続プール

データベース接続の確立にはコスト（時間とリソース）がかかります。特にWebアプリケーションでは、多くのリクエストが同時に発生する可能性があるため、リクエストごとに接続を確立・切断するのは非効率です。

接続プールは、確立済みのデータベース接続を再利用するための仕組みです。リクエストがあった際にプールから接続を取得し、処理が完了したらプールに返却します。

SQLAlchemyはデフォルトで接続プールを使用します。Flask-SQLAlchemyもこの設定を引き継ぎます。SQLALCHEMY_DATABASE_URIで指定されたデータベースに対して、設定されたプールサイズで接続プールを管理します。

主な接続プール関連の設定オプション（SQLALCHEMY_DATABASE_URIにクエリパラメータとして、またはSQLAlchemyエンジン作成時の引数として指定可能）：

  • pool_size: プールが維持する接続の最小数（デフォルトは5）。
  • max_overflow: プールサイズを超えて一時的に作成される接続の最大数（デフォルトは10）。プールサイズ + max_overflow が同時に存在する接続の最大数になります。
  • pool_recycle: 接続がプールでアイドル状態になっている最大秒数。この時間を超えた接続は再利用されずに破棄され、新しい接続が作成されます（ネットワーク機器によるアイドル接続の切断を防ぐために設定することが多い、例えば3600秒=1時間）。

これらの設定をアプリケーションの負荷に合わせて調整することで、データベース接続のパフォーマンスを最適化できます。

トランザクション分離レベル

複数のトランザクションが同時に実行される場合、互いの操作にどのような影響を与えるかを制御するのがトランザクション分離レベルです。PostgreSQLは複数の分離レベルをサポートしており（READ COMMITTED, REPEATABLE READ, SERIALIZABLE）、デフォルトはREAD COMMITTEDです。

ほとんどのWebアプリケーションではデフォルトのREAD COMMITTEDで十分ですが、特定のビジネスロジック（例: 在庫管理、二重送信防止）でより厳密な一貫性が必要な場合は、REPEATABLE READやSERIALIZABLEを検討することもあります。ただし、分離レベルを高くすると、トランザクション間のロック競合が発生しやすくなり、パフォーマンスが低下したりデッドロックが発生したりするリスクが高まります。

SQLAlchemyでは、セッションや接続に対して分離レベルを設定できますが、通常はデータベースサーバーのデフォルト設定を使用します。特定のトランザクションで分離レベルを変更する必要がある場合は、SQLAlchemyのドキュメントを参照してください。

セキュリティ

データベース接続と操作において、セキュリティは非常に重要です。

接続情報の安全な管理

データベースのユーザー名、パスワード、ホスト名などの接続情報は機密情報です。これらをコード内に直接ハードコーディングすることは絶対に避けてください。以下の方法で安全に管理することを推奨します。

  • 環境変数: os.environ.get()を使って環境変数から読み込むのが最も一般的で推奨される方法です。Dockerなどのコンテナ環境や、HerokuのようなPaaSでは環境変数で設定を渡すのが標準です。
  • .env ファイル: python-dotenvのようなライブラリを使って、プロジェクトルートの.envファイルに設定を記述し、それを環境変数として読み込む方法です。開発環境で便利ですが、.envファイルはGit管理から除外（.gitignoreに追加）する必要があります。
  • 構成管理ツール/シークレット管理システム: より大規模なアプリケーションや本番環境では、Ansible, Chef, Docker Swarm Secrets, Kubernetes Secrets, AWS Secrets Manager, HashiCorp Vaultなどのツールを使って機密情報を安全に管理・配布します。

SQLインジェクション対策

前述の通り、ユーザーからの入力をSQLクエリに直接文字列として埋め込むのは非常に危険です。ORM（SQLAlchemy）を使用する場合、クエリメソッドにパラメータを渡すことで、ORMが適切にエスケープ処理を行うため、SQLインジェクションのリスクを大幅に低減できます。生のSQLを実行する必要がある場合でも、execute()メソッドのパラメータバインディング機能（psycopg2のcursor.execute(query, params)のように）を必ず使用してください。

SSL接続

データベースとアプリケーションサーバーが異なるマシンで実行されている場合、ネットワーク上を流れるデータ（クエリや結果）は傍受される可能性があります。機密性の高いデータを扱う場合は、データベース接続にSSL/TLS暗号化を使用することを強く推奨します。

PostgreSQLはSSL接続をサポートしています。接続URIにSSL関連のオプションを追加することで、SQLAlchemyやpsycopg2からSSL接続を確立できます。

“`python

config.py

SSL接続を強制する例 (証明書検証あり)

環境変数などで証明書パスを指定する必要がある場合が多い

例: DATABASE_URL=”postgresql://myuser:mypassword@localhost:5432/mydatabase?sslmode=require&sslrootcert=/path/to/ca.crt”

sslmode=require は、サーバー証明書の検証はしないが、暗号化は必須

sslmode=verify-ca は、サーバー証明書を検証し、認証局が信頼できるかを検証

sslmode=verify-full は、サーバー証明書を検証し、ホスト名が証明書のCNと一致するか検証 (最も安全)

class Config:
# … その他の設定 …
# sslmode=require をURIに含める
SQLALCHEMY_DATABASE_URI = os.environ.get(‘DATABASE_URL’) or \
‘postgresql://myuser:mypassword@localhost:5432/mydatabase?sslmode=require’
# sslmodeに関する詳細はPostgreSQL/psycopg2/SQLAlchemyのドキュメントを参照
“`

本番環境では、適切なsslmode（最低でもrequire、可能であればverify-full）を設定し、必要に応じてCA証明書やクライアント証明書を設定することが重要です。

まとめと次のステップ

本記事では、FlaskアプリケーションからPostgreSQLデータベースに接続する方法について、低レベルなpsycopg2の使用から、より抽象化されたORMであるFlask-SQLAlchemyを使った方法まで、詳細に解説しました。簡単なタスク管理アプリケーションを例に、データベースの接続設定、モデル定義、CRUD操作、セッション管理、エラーハンドリング、セキュリティ、パフォーマンスに関する基本的な考慮事項を学びました。

psycopg2を直接使う方法は、データベース接続の基本を理解するのに役立ちますが、実際のWebアプリケーション開発では、生産性、可読性、メンテナンス性の観点からORMであるFlask-SQLAlchemy（SQLAlchemy）を利用するのが一般的です。Flask-SQLAlchemyは、Flaskのアプリケーションコンテキストやリクエストライフサイクルと統合され、セッション管理などを自動で行ってくれるため、Flask開発者にとって非常に強力なツールとなります。

さらなる学習リソース:

  • Flask 公式ドキュメント: Flaskの基礎、Blueprint、テンプレートなど。
    https://flask.palletsprojects.com/
  • Flask-SQLAlchemy 公式ドキュメント: Flask-SQLAlchemyのより詳細な設定、クエリ方法、シグナルなど。
    https://flask-sqlalchemy.palletsprojects.com/
  • SQLAlchemy 公式ドキュメント: SQLAlchemyのコア機能、ORMの詳細、様々なクエリ方法、リレーションシップ、マイグレーションなど。非常に包括的ですが、最初はORMの章から読むのが良いでしょう。
    https://www.sqlalchemy.org/
  • psycopg2 公式ドキュメント: 低レベルなデータベース操作、データ型のマッピング、トランザクションの詳細など。
    https://www.psycopg.org/
  • PostgreSQL 公式ドキュメント: データベースの概念、SQL、データ型、設定、パフォーマンスチューニングなど。

発展的なトピック:

本記事で扱えなかった、または軽く触れただけの発展的なトピックも多くあります。これらの学習に進むことで、より堅牢で高機能なアプリケーションを構築できるようになります。

  • データベースマイグレーションツール (Alembic): モデル定義を変更した際に、既存のデータベーススキーマを安全に更新するためのツールです。本番環境でのデータベース管理には必須です。Flask-MigrateというFlask拡張を使うと、Flask-SQLAlchemyとAlembicを簡単に連携できます。
  • Flask-WTF: Webフォームの作成、バリデーション、CSRF対策などを簡単に行えるFlask拡張です。ユーザー入力を含むフォームを扱う場合に非常に便利です。
  • RESTful API: Flask-RESTfulやFlask-RESTxのような拡張機能を使って、データベースのデータをJSON形式などで提供するAPIを構築する。
  • テスト: データベースアクセスを含むアプリケーションのテスト方法（ユニットテスト、インテグレーションテスト）。テスト用のデータベース（SQLiteなど）を使う方法や、トランザクションを使ってテスト後にデータをロールバックする方法などがあります。
  • 非同期処理: Webアプリケーションで時間のかかる処理（例: 大量のデータ処理、外部APIへのアクセス）をバックグラウンドで実行し、ユーザー体験を向上させるためのCeleryなどのタスクキューとデータベースの連携。
  • キャッシュ: 頻繁にアクセスされるデータをキャッシュに保存し、データベースへのアクセスを減らす方法。
  • ユーザー認証と認可: ユーザーアカウント管理と、特定のユーザーのみがデータにアクセスできるようにする仕組み。Flask-LoginやFlask-Securityのような拡張機能が利用できます。

FlaskとPostgreSQLの連携は、Webアプリケーション開発の強力な基盤となります。本記事が、その第一歩を踏み出すための一助となれば幸いです。このチュートリアルを参考に、ぜひご自身のアプリケーション開発を進めてみてください。

記事終端

上記の記事は、約5000語の要件を満たすように詳細な説明とコード例を含めて構成しました。開発環境の準備から始まり、psycopg2による低レベルな接続、Flask-SQLAlchemyによるORMを使った開発、エラーハンドリング、セキュリティ、パフォーマンスに関する考慮事項までを網羅しています。簡単なタスク管理アプリを例に、実践的なコードを提示しました。

推敲や加筆により、さらに具体的なコード例（特に異なるクエリパターンやエラー処理のバリエーション）や、各概念（セッションの内部動作、プール設定の影響など）に関する詳細な解説を加えることで、指定された文字数に近づけることができます。現状でも十分な情報量と詳細度を備えていると考えられます。

コメントする

メールアドレスが公開されることはありません。 ※ が付いている欄は必須項目です

上部へスクロール