Pythonでは、SQLite、MySQL、PostgreSQL、Oracleなどの異なるデータベースシステムを扱う際、DB-API 2.0 (PEP 249) という共通の仕様に基づいて実装されたドライバを使用します。これにより、データベース製品が変わっても、Python側のコード（接続、実行、取得、コミットなどの手順）はほとんど同じ書き方で統一できます。
ここでは、主要な操作メソッドを表にまとめ、MySQLドライバ（mysqlclient / MySQLdb）を使用した標準的な実装例を紹介します。
目次
コネクション（接続）関連の操作
データベースへの接続、トランザクション管理を行うためのメソッド一覧です。
| 対象 | メソッド/関数 | 処理内容 |
| モジュール | connect(...) | データベースへの接続を確立し、コネクションオブジェクトを返します。引数にはホスト名、ユーザー名、パスワードなどを指定します。 |
| コネクション | conn.cursor() | SQLコマンドを実行するためのカーソルオブジェクトを生成して返します。 |
| コネクション | conn.commit() | トランザクションを確定し、保留中の変更（INSERT, UPDATE, DELETE等）をデータベースに反映します。 |
| コネクション | conn.rollback() | トランザクションを取り消し、前回のコミット以降の変更を破棄して元の状態に戻します。エラー発生時の処理に使用します。 |
| コネクション | conn.close() | データベースへの接続を閉じます。リソース解放のため、処理終了後に必ず呼び出します。 |
カーソル（実行）関連の操作
SQLの実行や結果の取得を行うためのメソッド一覧です。
| 対象 | メソッド | 処理内容 |
| カーソル | cur.execute(sql, params) | SQL文を実行します。プレースホルダを使用することでSQLインジェクション対策が可能です。 |
| カーソル | cur.fetchone() | クエリ実行結果から次の1行を取得します。データがない場合は None を返します。 |
| カーソル | cur.fetchall() | クエリ実行結果の残りのすべての行をリスト形式で取得します。 |
| カーソル | cur.close() | カーソルを閉じ、リソースを解放します。 |
実装サンプル：MySQLドライバのAPIレベル確認と標準的なトランザクション処理
以下は、MySQL用ドライバである MySQLdb を使用したサンプルコードです。APIレベルを確認した後、アクセスログ管理システムを想定したトランザクション処理（例外発生時のロールバックを含む）を行います。
※実行には mysqlclient ライブラリのインストールと、稼働中のMySQLサーバーが必要です。
import MySQLdb
import sys
# データベース接続設定（環境に合わせて変更してください）
DB_CONFIG = {
'host': 'localhost',
'user': 'admin_user',
'passwd': 'secure_password',
'db': 'access_log_db',
'charset': 'utf8mb4'
}
def check_api_compliance():
"""ドライバのDB-API準拠レベルを確認する"""
# apilevel属性は、そのモジュールがサポートするDB-APIのバージョンを示します
# '2.0' と表示されれば、標準仕様に完全に準拠しています
print(f"MySQLdb API Level: {MySQLdb.apilevel}")
# スレッド安全性レベルの確認（1: モジュール共有不可, 2: モジュール共有可など）
print(f"Thread Safety Level: {MySQLdb.threadsafety}")
print("-" * 30)
def register_access_log(log_data):
"""
アクセスログをデータベースに登録する処理
コネクション管理とトランザクション（コミット/ロールバック）の実装例
"""
conn = None
try:
# 1. 接続の確立
print("Connecting to database...")
# 実際の接続処理（デモ環境がない場合は例外になるためコメントアウト等で調整してください）
# conn = MySQLdb.connect(**DB_CONFIG)
# ※解説用ダミーオブジェクト（実際は上記connectの結果を使用）
class MockConnection:
def cursor(self): return MockCursor()
def commit(self): print(">> Commit successful.")
def rollback(self): print(">> Rollback executed.")
def close(self): print(">> Connection closed.")
class MockCursor:
def execute(self, sql, params): print(f"Executing SQL: {sql} with {params}")
def close(self): print("Cursor closed.")
conn = MockConnection() # ダミー接続
# 2. カーソルの生成
cur = conn.cursor()
try:
# 3. SQLの実行
sql = "INSERT INTO server_logs (user_id, endpoint, status) VALUES (%s, %s, %s)"
# 複数件の処理を想定
for log in log_data:
cur.execute(sql, (log['id'], log['path'], log['code']))
# 意図的なエラー発生テスト（必要に応じてコメントアウト）
# raise Exception("Simulation of database error")
# 4. コミット（確定）
conn.commit()
print("All logs registered successfully.")
except Exception as e:
# 5. ロールバック（取り消し）
print(f"Error occurred: {e}")
conn.rollback()
finally:
# 6. カーソルのクローズ
cur.close()
except MySQLdb.Error as e:
print(f"Database connection failed: {e}")
finally:
# 7. 接続のクローズ（必ず実行されるようにfinallyブロックに記述）
if conn:
conn.close()
def main():
# DB-APIレベルの確認
check_api_compliance()
# テストデータ
new_logs = [
{'id': 101, 'path': '/api/login', 'code': 200},
{'id': 102, 'path': '/api/dashboard', 'code': 200},
{'id': 105, 'path': '/admin/settings', 'code': 403}
]
# メイン処理実行
register_access_log(new_logs)
if __name__ == "__main__":
main()
ポイント
コード内の try...except...finally ブロックに注目してください。データベース操作においては、途中でエラーが発生した場合にデータの不整合を防ぐため、rollback() を呼び出す構成にすることが重要です。また、finally ブロックで確実に close() を呼び出し、サーバーへの接続リソースを解放します。
