はじめに

Pythonでデータベースを操作する機会は多く、特にMySQLとSQL Serverは多くのシステムで採用されている主要なリレーショナルデータベースです。本稿では、pymysqlとpymssqlという2つのライブラリを使用して、データベースの基本的な操作から実践的な応用までを体系的に解説します。

1. 必要なライブラリのインストール

データベース接続を行う前に、必要なライブラリをpipでインストールします。

pip install pymysql
pip install pymssql

SQL Server用のpymssqlは、FreeTDSライブラリに依存しているため、Windows環境ではVisual C++ Build Toolsのインストールが必要となる場合があります。

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

pymysqlを使用してMySQLデータベースに接続する基本的なパターンを以下に示します。

import pymysql

# データベース接続の確立
db_connection = pymysql.connect(
    host='localhost',
    user='db_user',
    password='secure_password',
    db='sample_database',
    charset='utf8mb4',
    port=3306,
    cursorclass=pymysql.cursors.DictCursor
)

try:
    with db_connection.cursor() as cursor:
        # SQLクエリの実行
        query = "SELECT id, name, created_at FROM users WHERE status = %s"
        cursor.execute(query, ('active',))
        
        # 結果の取得
        records = cursor.fetchall()
        for record in records:
            print(f"ID: {record['id']}, Name: {record['name']}")

finally:
    db_connection.close()

このコードでは、接続時にcursorclass=pymysql.cursors.DictCursorを指定することで、結果を辞書形式で取得できるようにしています。これにより、列名でのアクセスが可能となり、コードの可読性が向上します。

3. SQL Serverデータベースへの接続

SQL Serverに接続する場合は、pymssqlを使用します。接続の確立方法はMySQLとよく似ていますが、若干異なるパラメータがあります。

import pymssql

# SQL Serverへの接続
server_connection = pymssql.connect(
    server='localhost',
    user='sa',
    password='YourPassword123',
    database='SampleDB'
)

try:
    with server_connection.cursor() as cursor:
        # データの取得
        cursor.execute("SELECT product_id, product_name, price FROM products")
        results = cursor.fetchall()
        
        for row in results:
            print(f"Product: {row[1]}, Price: ${row[2]}")

finally:
    server_connection.close()

SQL Serverでは、Windows認証を使用した接続もサポートされています。その場合はtrusted_connection=Trueパラメータを追加し、ユーザー名とパスワードを省略します。

4. データの挿入処理

データベースへのデータ挿入は、INSERT文を使用して行います。Pythonでは、プレースホルダを使用したパラメータ化クエリを実行することで、SQLインジェクション攻撃を防止できます。

import pymysql

# 接続の確立
connection = pymysql.connect(
    host='localhost',
    user='db_user',
    password='password',
    db='sample_database'
)

with connection.cursor() as cursor:
    # データの挿入
    insert_statement = "INSERT INTO employees (name, department, salary) VALUES (%s, %s, %s)"
    employee_data = ('田中太郎', '開発部', 550000)
    cursor.execute(insert_statement, employee_data)
    
    # 複数件の挿入
    additional_data = [
        ('佐藤花子', '人事部', 480000),
        ('鈴木次郎', '営業部', 520000)
    ]
    cursor.executemany(insert_statement, additional_data)

# トランザクションのコミット
connection.commit()
connection.close()

executemany()メソッドを使用すると、複数件のデータを効率的に一括挿入できます。挿入操作後は、必ずcommit()を呼び出してトランザクションを確定させることを忘れないでください。

5. データの更新処理

既存データの更新には、UPDATE文を使用します。WHERE句を適切に設定することで、特定の条件に合致するレコードのみを更新できます。

import pymysql

db_connection = pymysql.connect(
    host='localhost',
    user='db_user',
    password='password',
    db='sample_database'
)

with db_connection.cursor() as cursor:
    # データの更新
    update_sql = "UPDATE users SET last_login = %s WHERE username = %s"
    import datetime
    current_time = datetime.datetime.now()
    cursor.execute(update_sql, (current_time, 'test_user'))
    
    print(f"更新されたレコード数: {cursor.rowcount}")

db_connection.commit()
db_connection.close()

cursor.rowcountプロパティは、直近の操作で影響を受けた行数を返します。これにより、更新処理が正常に実行されたことを確認できます。

6. データの検索と結果の処理

SELECT文によるデータ検索は、データベース操作の中で最も頻繁に行われる処理です。largeテーブルを扱う場合は、fetchone()を使用してメモリ効率の良い処理を検討してください。

import pymysql

connection = pymysql.connect(
    host='localhost',
    user='db_user',
    password='password',
    db='sample_database'
)

try:
    with connection.cursor(pymysql.cursors.DictCursor) as cursor:
        # 複雑な検索クエリの実行
        search_query = """
            SELECT 
                u.id, u.name, u.email, COUNT(o.id) as order_count
            FROM users u
            LEFT JOIN orders o ON u.id = o.user_id
            WHERE u.created_at >= %s
            GROUP BY u.id
            HAVING order_count > %s
        """
        cursor.execute(search_query, ('2023-01-01', 5))
        
        # 全結果の取得
        user_data = cursor.fetchall()
        
        for user in user_data:
            print(f"ユーザー: {user['name']}, 注文数: {user['order_count']}")

finally:
    connection.close()

JOIN操作やGROUP BY、 aggregate関数を使用した複雑なクエリも、同様の方法で実行できます。结果をfetchall()で一度に取得するか、fetchone()で一件ずつ処理するかは、データ量に応じて適切に選択してください。

7. 例外処理の実装

データベース操作では、ネットワーク障害やクエリエラーなど、様々な例外が発生する可能性があります。適切な例外処理を実装することで、アプリケーションの堅牢性を確保できます。

import pymysql
from pymysql import OperationalError, ProgrammingError

def fetch_user_data(user_id):
    try:
        connection = pymysql.connect(
            host='localhost',
            user='db_user',
            password='password',
            db='sample_database'
        )
        
        with connection.cursor() as cursor:
            cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
            return cursor.fetchone()
            
    except OperationalError as e:
        print(f"接続エラーが発生しました: {e}")
        return None
    except ProgrammingError as e:
        print(f"SQLエラーが発生しました: {e}")
        return None
    except Exception as e:
        print(f"予期しないエラー: {e}")
        return None
    finally:
        if 'connection' in locals():
            connection.close()

この例では、接続エラーとSQLエラーを区別して処理しています。finallyブロックを使用することで、エラーの有無にかかわらず確実に接続をクローズできます。

8. パラメータ化クエリの活用

パラメータ化クエリは、SQLインジェクションを防止するための重要な手法です。ユーザー入力を直接SQL문에連結するのではなく、必ずプレースホルダを使用してください。

import pymysql

connection = pymysql.connect(
    host='localhost',
    user='db_user',
    password='password',
    db='sample_database'
)

# 危険な例（決して行わないでください）
# user_input = "'; DROP TABLE users; --"
# cursor.execute(f"SELECT * FROM users WHERE name = '{user_input}'")

# 安全なパラメータ化クエリ
with connection.cursor() as cursor:
    search_term = "管理者"  # ユーザーからの入力値
    status_filter = "active"
    
    safe_query = "SELECT * FROM users WHERE name LIKE %s AND status = %s"
    cursor.execute(safe_query, (f'%{search_term}%', status_filter))
    
    results = cursor.fetchall()
    print(f"検索ヒット数: {len(results)}")

connection.close()

LIKE句での検索では、ワイルドカード文字（%）を含む値をプレースホルダで渡す必要があります。f'%{search_term}%'のようにPython側でワイルドカードを追加する方法と、SQL内でCONCAT('%', %s, '%')を使用する方法があります。

9. コンテキストマネージャの活用

Pythonのwith文（コンテキストマネージャ）を使用することで、接続とカーソルの管理を自動化できます。これにより、コードの安全性と可読性が向上します。

import pymysql

def get_user_by_email(email):
    """指定されたメールアドレスを持つユーザーを取得"""
    with pymysql.connect(
        host='localhost',
        user='db_user',
        password='password',
        db='sample_database'
    ) as connection:
        with connection.cursor(pymysql.cursors.DictCursor) as cursor:
            cursor.execute("SELECT * FROM users WHERE email = %s", (email,))
            return cursor.fetchone()

# 使用例
user = get_user_by_email('user@example.com')
if user:
    print(f"ユーザー情報: {user['name']}")
else:
    print("ユーザーが見つかりません")

コンテキストマネージャを使用した場合、ブロックの終了時に自動的にcommit()またはrollback()が呼ばれ、接続もクローズされます。明示的にcommit()を呼び出さなかった場合、暗黙的にロールバックが行われる点に注意が必要です。

10. バルクインサートの実装

大量のデータを挿入する場合は、個別にINSERT文を実行するよりも、バルク插入を使用した方がパフォーマンスが大幅に向上します。

import pymysql
from datetime import datetime

connection = pymysql.connect(
    host='localhost',
    user='db_user',
    password='password',
    db='sample_database'
)

# 挿入対象のデータリスト
bulk_data = [
    ('製品A', 1000, datetime.now(), 1),
    ('製品B', 2000, datetime.now(), 1),
    ('製品C', 1500, datetime.now(), 2),
    ('製品D', 3000, datetime.now(), 1),
    ('製品E', 2500, datetime.now(), 3),
]

with connection.cursor() as cursor:
    bulk_query = """
        INSERT INTO products (name, price, created_at, category_id)
        VALUES (%s, %s, %s, %s)
    """
    cursor.executemany(bulk_query, bulk_data)

connection.commit()
connection.close()

print(f"{len(bulk_data)}件のデータを挿入しました")

数百から数千件のデータを扱う場合、バルク挿入を使用することでネットワークオーバーヘッドを大幅に削減できます。ただし、挿入するデータ量が多すぎる場合は、小さなバッチに分割して処理することをお勧めします。

11. SQLAlchemy ORMの使用

直接SQLを実行する代わりに、ORM（Object-Relational Mapping）フレームワークを使用することで、より抽象的でメンテナンス性の高いコードを書けます。SQLAlchemyはPythonで最も広く使用されているORMの一つです。

from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime

Base = declarative_base()

class Product(Base):
    __tablename__ = 'products'
    
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(100), nullable=False)
    price = Column(Integer, nullable=False)
    created_at = Column(DateTime, default=datetime.now)

# データベースエンジンの作成
engine = create_engine(
    'mysql+pymysql://db_user:password@localhost/sample_database',
    echo=False
)

# テーブルの作成
Base.metadata.create_all(engine)

# セッションの作成
Session = sessionmaker(bind=engine)
session = Session()

# データのクエリ
results = session.query(Product).filter(Product.price > 1000).all()

for product in results:
    print(f"製品名: {product.name}, 価格: {product.price}")

session.close()

ORMを使用することで、データベースの詳細を意識せずにオブジェクト指向的なコードでデータ操作を行えます。ただし、パフォーマンスが重要な処理では、生のSQLの方が効率的な場合もあることを覚えておいてください。

12. トランザクション管理

トランザクションは、複数のデータベース操作を 하나의単位として扱います。すべての操作が成功した場合はコミットし、一つでも失敗した場合はロールバックすることで、データの整合性を保つことができます。

import pymysql

def transfer_funds(from_account, to_account, amount):
    """口座間の資金移転処理"""
    connection = pymysql.connect(
        host='localhost',
        user='db_user',
        password='password',
        db='sample_database'
    )
    
    try:
        with connection.cursor() as cursor:
            # 残高確認
            cursor.execute("SELECT balance FROM accounts WHERE id = %s", (from_account,))
            balance = cursor.fetchone()
            
            if not balance or balance[0] < amount:
                raise ValueError("残高不足または口座が存在しません")
            
            # 出金
            cursor.execute(
                "UPDATE accounts SET balance = balance - %s WHERE id = %s",
                (amount, from_account)
            )
            
            # 入金
            cursor.execute(
                "UPDATE accounts SET balance = balance + %s WHERE id = %s",
                (amount, to_account)
            )
        
        # 両方の更新が成功した場合のみコミット
        connection.commit()
        print("資金移転が完了しました")
        
    except Exception as e:
        connection.rollback()
        print(f"エラーによりロールバックされました: {e}")
        raise
    finally:
        connection.close()

この例では、二つのUPDATE文を一つのトランザクションとして扱っています。いずれか一方でも失敗した場合、すべての変更が取り消されるため、不整合な状態を防ぐことができます。

13. 接続プールによる効率化

高負荷環境では、データベース接続の確立と破棄のオーバーヘッドが問題になることがあります。接続プールを使用することで、接続を再利用しパフォーマンスを向上させます。

from DBUtils.PooledDB import PooledDB
import pymysql

# 接続プールの設定
connection_pool = PooledDB(
    creator=pymysql,
    maxconnections=10,
    mincached=2,
    maxcached=5,
    blocking=True,
    host='localhost',
    user='db_user',
    password='password',
    database='sample_database',
    port=3306,
    charset='utf8mb4'
)

# プールから接続を取得して使用
def execute_query(query, params=None):
    conn = connection_pool.connection()
    try:
        with conn.cursor() as cursor:
            if params:
                cursor.execute(query, params)
            else:
                cursor.execute(query)
            return cursor.fetchall()
    finally:
        conn.close()

# 使用例
results = execute_query("SELECT COUNT(*) FROM users")
print(f"ユーザー総数: {results[0][0]}")

接続プールを適切に設定することで、同時接続ユーザー数に応じたスケーラビリティを確保できます。maxconnectionsパラメータはプール全体で管理される最大接続数を制御します。

おわりに

本稿では、Pythonを使用してMySQLとSQL Server两大データベースを操作するための実践的な手法を解説しました。基本的な接続方法から、パラメータ化クエリによるセキュリティ強化、トランザクション管理、ORMの活用、そして接続プールによるパフォーマンス最適化まで涵盖しました。

実際の開発では、扱うデータの特性やシステムの要件に応じて、適切な手法を選択することが重要です。セキュリティ、パフォーマンス、コードの可読性のバランスを考慮しながら、本稿の知識を活用して効果的なデータベースアプリケーションを構築してください。