👌

PythonプロジェクトでSQLiteデータベースを作成・管理する方法

2025/07/10に公開

Pythonプロジェクトでデータベースを使いたいけど、PostgreSQLやMySQLの環境構築が面倒だと思ったことはありませんか？
SQLiteを使えば、ファイルベースのデータベースで簡単に始められ、DBeaverで視覚的に管理できます。

今回は、PoetryプロジェクトでSQLiteデータベースを作成し、DBeaverで確認する方法を紹介します。将来的にRDBやAWSに移行できる構成になっています。

🚨 問題：データベース環境構築の課題

従来のデータベース開発では、以下のような課題がありました：

環境構築の複雑さ: PostgreSQLやMySQLのインストール・設定が面倒
Dockerの学習コスト: コンテナ技術の理解が必要
チーム開発の困難: 環境の統一が難しい
開発・本番環境の違い: ローカルとサーバーで設定が異なる
初心者のハードル: データベースの概念理解が大変

🆕 解決策：SQLite + Python + DBeaver

SQLiteを使うことで、以下のメリットがあります：

SQLiteの特徴

ファイルベース: 単一ファイルでデータベース管理
設定不要: インストールや設定が不要
軽量: リソース消費が少ない
移植性: ファイルをコピーするだけで移行可能
標準対応: Pythonに標準で含まれている

DBeaverの特徴

無料: オープンソースのデータベース管理ツール
多様対応: SQLite、PostgreSQL、MySQLなど多数対応
視覚的: GUIでデータベースを操作
初心者向け: 直感的なインターフェース

🛠 手順1：プロジェクトの準備

まずは、PoetryプロジェクトでSQLiteを使うための環境を準備します。

1. Poetryプロジェクトの作成

# 新しいプロジェクトを作成
poetry new my-database-app
cd my-database-app

# プロジェクト内仮想環境を有効化
poetry config virtualenvs.in-project true --local

# 必要な依存関係を追加
poetry add sqlite3  # Python標準ライブラリなので不要
poetry add --group dev pytest black flake8

2. プロジェクト構造

my-database-app/
├── pyproject.toml
├── my_database_app/
│   ├── __init__.py
│   ├── database/
│   │   ├── __init__.py
│   │   ├── manager.py      # データベース管理クラス
│   │   └── schema.sql      # スキーマ定義
│   ├── models/
│   │   ├── __init__.py
│   │   └── user.py         # ユーザーモデル
│   └── utils/
│       ├── __init__.py
│       └── config.py       # 設定管理
├── scripts/
│   └── create_db.py        # データベース作成スクリプト
└── tests/
    └── test_database.py

🔄 手順2：データベース管理クラスの作成

SQLiteデータベースを管理するクラスを作成します。

1. データベース管理クラス

# my_database_app/database/manager.py
import sqlite3
import logging
from pathlib import Path
from typing import Optional, List, Dict, Any

class DatabaseManager:
    """SQLiteデータベース管理クラス"""
    
    def __init__(self, db_path: Optional[str] = None):
        """
        データベースマネージャーを初期化
        
        Args:
            db_path: データベースファイルのパス（Noneの場合はデフォルト）
        """
        self.db_path = db_path or "data/app.db"
        self.connection: Optional[sqlite3.Connection] = None
        
        # データベースディレクトリを作成
        Path(self.db_path).parent.mkdir(parents=True, exist_ok=True)
        
        # ログ設定
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)
    
    def connect(self) -> sqlite3.Connection:
        """
        データベースに接続
        
        Returns:
            sqlite3.Connection: データベース接続オブジェクト
        """
        if self.connection is None:
            self.connection = sqlite3.connect(self.db_path)
            # 外部キー制約を有効化
            self.connection.execute("PRAGMA foreign_keys = ON")
            # 結果を辞書形式で取得
            self.connection.row_factory = sqlite3.Row
            
        return self.connection
    
    def disconnect(self) -> None:
        """データベース接続を閉じる"""
        if self.connection:
            self.connection.close()
            self.connection = None
    
    def execute_query(self, query: str, params: tuple = ()) -> List[Dict[str, Any]]:
        """
        SQLクエリを実行して結果を取得
        
        Args:
            query: SQLクエリ
            params: クエリパラメータ
            
        Returns:
            List[Dict[str, Any]]: クエリ結果
        """
        conn = self.connect()
        cursor = conn.cursor()
        
        try:
            cursor.execute(query, params)
            
            if query.strip().upper().startswith('SELECT'):
                # SELECT文の場合は結果を返す
                return [dict(row) for row in cursor.fetchall()]
            else:
                # INSERT/UPDATE/DELETE文の場合はコミット
                conn.commit()
                return []
                
        except sqlite3.Error as e:
            self.logger.error(f"クエリ実行エラー: {e}")
            conn.rollback()
            raise
        finally:
            cursor.close()
    
    def create_schema(self) -> None:
        """データベーススキーマを作成"""
        schema_path = Path(__file__).parent / "schema.sql"
        
        if not schema_path.exists():
            raise FileNotFoundError(f"スキーマファイルが見つかりません: {schema_path}")
        
        # スキーマファイルを読み込み
        with open(schema_path, 'r', encoding='utf-8') as f:
            schema_sql = f.read()
        
        # スキーマを実行
        self.execute_query(schema_sql)
        self.logger.info("データベーススキーマを作成しました")
    
    def test_connection(self) -> None:
        """データベース接続をテスト"""
        try:
            result = self.execute_query("SELECT sqlite_version() as version")
            version = result[0]['version']
            self.logger.info(f"SQLite接続成功: バージョン {version}")
        except Exception as e:
            self.logger.error(f"データベース接続テスト失敗: {e}")
            raise
    
    def insert_sample_data(self) -> None:
        """サンプルデータを挿入"""
        # ユーザーデータ
        users_data = [
            ("田中太郎", "tanaka@example.com", "active"),
            ("佐藤花子", "sato@example.com", "active"),
            ("鈴木一郎", "suzuki@example.com", "inactive"),
        ]
        
        for name, email, status in users_data:
            self.execute_query(
                "INSERT INTO users (name, email, status) VALUES (?, ?, ?)",
                (name, email, status)
            )
        
        # カテゴリデータ
        categories_data = [
            ("技術", "プログラミングやIT関連"),
            ("デザイン", "UI/UXデザイン関連"),
            ("ビジネス", "経営・マーケティング関連"),
        ]
        
        for name, description in categories_data:
            self.execute_query(
                "INSERT INTO categories (name, description) VALUES (?, ?)",
                (name, description)
            )
        
        # 記事データ
        articles_data = [
            ("Python入門", "Pythonの基礎を学ぼう", 1, 1),
            ("SQLiteの使い方", "SQLiteでデータベースを作成", 1, 1),
            ("デザインの基本", "良いデザインの原則", 2, 2),
        ]
        
        for title, content, user_id, category_id in articles_data:
            self.execute_query(
                "INSERT INTO articles (title, content, user_id, category_id) VALUES (?, ?, ?, ?)",
                (title, content, user_id, category_id)
            )
        
        self.logger.info("サンプルデータを挿入しました")

2. スキーマ定義ファイル

-- my_database_app/database/schema.sql

-- =============================================================================
-- ユーザーテーブル定義
-- =============================================================================
CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,  -- ユーザーID（自動採番）
    name TEXT NOT NULL,                    -- ユーザー名
    email TEXT NOT NULL UNIQUE,            -- メールアドレス（重複不可）
    status TEXT NOT NULL DEFAULT 'active' CHECK (status IN ('active', 'inactive')),  -- ステータス
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,  -- 作成日時
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP   -- 更新日時
);

-- =============================================================================
-- カテゴリテーブル定義
-- =============================================================================
CREATE TABLE IF NOT EXISTS categories (
    id INTEGER PRIMARY KEY AUTOINCREMENT,  -- カテゴリID（自動採番）
    name TEXT NOT NULL UNIQUE,             -- カテゴリ名（重複不可）
    description TEXT,                      -- カテゴリの説明
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP  -- 作成日時
);

-- =============================================================================
-- 記事テーブル定義
-- =============================================================================
CREATE TABLE IF NOT EXISTS articles (
    id INTEGER PRIMARY KEY AUTOINCREMENT,  -- 記事ID（自動採番）
    title TEXT NOT NULL,                   -- 記事タイトル
    content TEXT NOT NULL,                 -- 記事内容
    user_id INTEGER NOT NULL,              -- 投稿者ID（外部キー）
    category_id INTEGER NOT NULL,          -- カテゴリID（外部キー）
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,  -- 作成日時
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,  -- 更新日時
    
    -- 外部キー制約
    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
    FOREIGN KEY (category_id) REFERENCES categories(id) ON DELETE CASCADE
);

-- =============================================================================
-- インデックス作成（検索性能向上）
-- =============================================================================
CREATE INDEX IF NOT EXISTS idx_users_email ON users(email);           -- メールアドレス検索用
CREATE INDEX IF NOT EXISTS idx_users_status ON users(status);         -- ステータス検索用
CREATE INDEX IF NOT EXISTS idx_articles_user_id ON articles(user_id); -- ユーザー別記事検索用
CREATE INDEX IF NOT EXISTS idx_articles_category_id ON articles(category_id); -- カテゴリ別記事検索用
CREATE INDEX IF NOT EXISTS idx_articles_created_at ON articles(created_at);   -- 日付順検索用

-- =============================================================================
-- トリガー作成（updated_at自動更新）
-- =============================================================================
CREATE TRIGGER IF NOT EXISTS update_users_updated_at 
    AFTER UPDATE ON users
    FOR EACH ROW
BEGIN
    UPDATE users SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id;
END;

CREATE TRIGGER IF NOT EXISTS update_articles_updated_at 
    AFTER UPDATE ON articles
    FOR EACH ROW
BEGIN
    UPDATE articles SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id;
END;

✅ 手順3：データベース作成スクリプト

CLIスクリプトを作成して、データベースの初期化を簡単に行えるようにします。

#!/usr/bin/env python3
# scripts/create_db.py

"""
データベース作成CLIスクリプト

このスクリプトは、アプリケーションの初期セットアップを行います。
必要なディレクトリの作成、データベーススキーマの適用、接続テストを実行します。

使用方法:
    poetry run python scripts/create_db.py
    または
    make create-db
"""

import sys
from pathlib import Path
from typing import NoReturn

def main() -> None:
    """
    メイン関数
    
    データベース作成の全プロセスを実行します。
    エラーが発生した場合は適切なエラーメッセージを表示して終了します。
    """
    try:
        # プロジェクトルートをPythonパスに追加
        project_root = Path(__file__).parent.parent
        sys.path.insert(0, str(project_root))
        
        # 遅延import（Poetryのパッケージ設定により自動的に利用可能）
        from my_database_app.database.manager import DatabaseManager
        
        print("🗄️  データベース作成スクリプト")
        print("=" * 50)
        
        # データベース管理クラス初期化
        db_manager = DatabaseManager()
        
        print(f"📊 データベースパス: {db_manager.db_path}")
        print("🔧 データベーススキーマを作成中...")
        
        # スキーマ作成（テーブル作成）
        db_manager.create_schema()
        
        print("🧪 接続テストを実行中...")
        # 接続テスト
        db_manager.test_connection()
        
        print("📝 サンプルデータを挿入中...")
        # サンプルデータ挿入
        db_manager.insert_sample_data()
        
        print("✅ データベース作成完了！")
        print(f"📁 データベースファイル: {db_manager.db_path}")
        print("💡 DBeaverで接続してデータを確認できます")
        
    except ImportError as e:
        print(f"\n❌ モジュールのインポートエラー: {e}")
        print("💡 解決方法:")
        print("   1. poetry install を実行してください")
        print("   2. poetry shell で仮想環境を起動してください")
        sys.exit(1)
    except FileNotFoundError as e:
        print(f"\n❌ ファイル・ディレクトリが見つかりません: {e}")
        print("💡 解決方法:")
        print("   1. プロジェクトルートディレクトリで実行してください")
        sys.exit(1)
    except Exception as e:
        print(f"\n❌ 予期しないエラーが発生しました: {e}")
        print("💡 解決方法:")
        print("   1. エラーメッセージを確認してください")
        print("   2. 必要に応じて管理者に連絡してください")
        sys.exit(1)
    finally:
        # データベース接続を閉じる
        if 'db_manager' in locals():
            db_manager.disconnect()

def error_exit(message: str, exit_code: int = 1) -> NoReturn:
    """
    エラーメッセージを表示して終了
    
    Args:
        message: エラーメッセージ
        exit_code: 終了コード
    """
    print(f"\n❌ {message}")
    sys.exit(exit_code)

if __name__ == "__main__":
    main()

📊 手順4：DBeaverでの接続・確認

DBeaverを使ってデータベースを視覚的に管理します。

1. DBeaverのインストール

# macOS（Homebrew）
brew install --cask dbeaver-community

# Windows
# https://dbeaver.io/download/ からダウンロード

# Linux
# https://dbeaver.io/download/ からダウンロード

2. SQLiteデータベースへの接続

DBeaverを起動
新しい接続を作成
- 「新しい接続」ボタンをクリック
- 「SQLite」を選択
- 「次へ」をクリック
接続設定
- データベースファイル: プロジェクトのdata/app.dbを選択
- 接続名: MyApp Databaseなど分かりやすい名前
接続テスト
- 「接続テスト」ボタンをクリック
- 成功したら「完了」をクリック

3. データベースの確認

3.1 テーブル構造の確認

-- テーブル一覧を表示
SELECT name FROM sqlite_master WHERE type='table';

-- ユーザーテーブルの構造を確認
PRAGMA table_info(users);

-- カテゴリテーブルの構造を確認
PRAGMA table_info(categories);

-- 記事テーブルの構造を確認
PRAGMA table_info(articles);

3.2 データの確認

-- ユーザーデータを確認
SELECT * FROM users;

-- カテゴリデータを確認
SELECT * FROM categories;

-- 記事データを確認（JOINで関連データも表示）
SELECT 
    a.title,
    a.content,
    u.name as author,
    c.name as category,
    a.created_at
FROM articles a
JOIN users u ON a.user_id = u.id
JOIN categories c ON a.category_id = c.id
ORDER BY a.created_at DESC;

3.3 データの追加・編集

DBeaverのGUIで以下の操作が可能です：

データの追加: テーブルを右クリック → 「データを編集」
SQL実行: SQLエディタでクエリを実行
データのエクスポート: CSV、JSON、Excel形式で出力
スキーマの確認: テーブル構造を視覚的に確認

🔄 手順5：Pythonアプリケーションでの使用例

実際のアプリケーションでデータベースを使用する例を紹介します。

1. ユーザーモデル

# my_database_app/models/user.py
from dataclasses import dataclass
from datetime import datetime
from typing import Optional, List

@dataclass
class User:
    """ユーザーモデル"""
    id: Optional[int]
    name: str
    email: str
    status: str
    created_at: datetime
    updated_at: datetime
    
    @classmethod
    def from_dict(cls, data: dict) -> 'User':
        """辞書からUserオブジェクトを作成"""
        return cls(
            id=data.get('id'),
            name=data['name'],
            email=data['email'],
            status=data['status'],
            created_at=datetime.fromisoformat(data['created_at']),
            updated_at=datetime.fromisoformat(data['updated_at'])
        )

class UserRepository:
    """ユーザーリポジトリクラス"""
    
    def __init__(self, db_manager):
        self.db_manager = db_manager
    
    def get_all(self) -> List[User]:
        """全ユーザーを取得"""
        query = "SELECT * FROM users ORDER BY created_at DESC"
        results = self.db_manager.execute_query(query)
        return [User.from_dict(row) for row in results]
    
    def get_by_id(self, user_id: int) -> Optional[User]:
        """IDでユーザーを取得"""
        query = "SELECT * FROM users WHERE id = ?"
        results = self.db_manager.execute_query(query, (user_id,))
        return User.from_dict(results[0]) if results else None
    
    def create(self, name: str, email: str, status: str = 'active') -> User:
        """新しいユーザーを作成"""
        query = """
            INSERT INTO users (name, email, status) 
            VALUES (?, ?, ?)
        """
        self.db_manager.execute_query(query, (name, email, status))
        
        # 作成されたユーザーを取得
        query = "SELECT * FROM users WHERE email = ?"
        results = self.db_manager.execute_query(query, (email,))
        return User.from_dict(results[0])
    
    def update_status(self, user_id: int, status: str) -> bool:
        """ユーザーのステータスを更新"""
        query = "UPDATE users SET status = ? WHERE id = ?"
        self.db_manager.execute_query(query, (status, user_id))
        return True

2. アプリケーションでの使用例

# my_database_app/main.py
from database.manager import DatabaseManager
from models.user import UserRepository

def main():
    """メインアプリケーション"""
    # データベースマネージャーを初期化
    db_manager = DatabaseManager()
    
    # ユーザーリポジトリを作成
    user_repo = UserRepository(db_manager)
    
    try:
        # 全ユーザーを取得
        users = user_repo.get_all()
        print(f"登録ユーザー数: {len(users)}")
        
        for user in users:
            print(f"- {user.name} ({user.email}) - {user.status}")
        
        # 新しいユーザーを作成
        new_user = user_repo.create("山田次郎", "yamada@example.com")
        print(f"新しいユーザーを作成: {new_user.name}")
        
    finally:
        # データベース接続を閉じる
        db_manager.disconnect()

if __name__ == "__main__":
    main()

📊 トラブルシューティング

よくある問題とその解決方法を紹介します。

1. データベースファイルが見つからない

# データベースファイルの場所を確認
find . -name "*.db" -type f

# データベースディレクトリを作成
mkdir -p data

2. DBeaverで接続できない

# データベースファイルの権限を確認
ls -la data/app.db

# 権限を修正
chmod 644 data/app.db

3. 外部キー制約エラー

-- 外部キー制約を確認
PRAGMA foreign_keys;

-- 外部キー制約を有効化
PRAGMA foreign_keys = ON;

4. データベースのバックアップ・復元

# バックアップ
cp data/app.db data/app.db.backup

# 復元
cp data/app.db.backup data/app.db

🚀 次のステップ：本格的なRDBへの移行

SQLiteから本格的なRDBへの移行方法を紹介します。

1. PostgreSQLへの移行

# requirements.txt または pyproject.toml
psycopg2-binary==2.9.5  # PostgreSQL接続用

# データベースマネージャーの修正例
import psycopg2
from psycopg2.extras import RealDictCursor

class PostgreSQLManager:
    def __init__(self, connection_string: str):
        self.connection_string = connection_string
    
    def connect(self):
        return psycopg2.connect(
            self.connection_string,
            cursor_factory=RealDictCursor
        )

2. AWS RDSへの移行

# AWS RDS接続例
import boto3
import psycopg2

class AWSRDSManager:
    def __init__(self, host: str, port: int, database: str, 
                 username: str, password: str):
        self.host = host
        self.port = port
        self.database = database
        self.username = username
        self.password = password
    
    def connect(self):
        return psycopg2.connect(
            host=self.host,
            port=self.port,
            database=self.database,
            user=self.username,
            password=self.password
        )

3. 移行スクリプト

# migration_script.py
import sqlite3
import psycopg2
from psycopg2.extras import RealDictCursor

def migrate_sqlite_to_postgresql(sqlite_path: str, pg_connection_string: str):
    """SQLiteからPostgreSQLにデータを移行"""
    
    # SQLiteからデータを読み込み
    sqlite_conn = sqlite3.connect(sqlite_path)
    sqlite_conn.row_factory = sqlite3.Row
    
    # PostgreSQLに接続
    pg_conn = psycopg2.connect(pg_connection_string)
    
    try:
        # ユーザーデータを移行
        users = sqlite_conn.execute("SELECT * FROM users").fetchall()
        for user in users:
            pg_conn.execute(
                "INSERT INTO users (name, email, status) VALUES (%s, %s, %s)",
                (user['name'], user['email'], user['status'])
            )
        
        pg_conn.commit()
        print("データ移行が完了しました")
        
    finally:
        sqlite_conn.close()
        pg_conn.close()

🧭 おわりに

SQLiteを使うことで、簡単にデータベース開発を始めることができます。

学習のポイント

SQLiteの基本: ファイルベースのデータベースの理解
SQLの基礎: テーブル作成、データ操作、JOIN
Python連携: データベース操作の自動化
DBeaver活用: 視覚的なデータベース管理
移行戦略: 将来的なスケールアップの準備

推奨する学習順序

SQLiteの基本操作 ← この記事で学習
DBeaverでの管理 ← この記事で学習
Pythonアプリケーションでの活用 ← この記事で学習
PostgreSQLへの移行 ← 次のステップ
AWS RDSでの運用 ← 本格運用

🚀 Google Colabでの動作確認

この記事の内容を実際に試してみたい方は、以下のGoogle Colabノートブックをご利用ください：

SQLiteデータベース作成デモ - Google Colab

このノートブックでは、以下の5つのステップでSQLiteデータベースの基本操作を体験できます：

SQLiteデータベースの作成 - メモリ上またはファイルベースでデータベースを作成
テーブルの作成 - ユーザー、カテゴリ、記事の3つのテーブルを作成
サンプルデータの挿入 - 実際のデータを挿入してテーブルを活用
データの取得・表示 - SELECT文とJOINを使ったデータ取得
リソースの解放 - 適切な接続終了処理

Google Colabを利用すれば、ローカル環境のセットアップなしで即座にSQLiteデータベースの動作を確認できます。各セルを順番に実行することで、データベース操作の流れを理解できます。

SQLiteは初心者でも簡単に始められ、本格的なアプリケーション開発の第一歩として最適です。ぜひ試してみてください！