# ===== to_sqlの基本 ===== import pandas as pd from sqlalchemy import create_engine # サンプルデータ df = pd.DataFrame({ ‘商品ID’: [1, 2, 3, 4, 5], ‘商品名’: [‘りんご’, ‘バナナ’, ‘みかん’, ‘ぶどう’, ‘いちご’], ‘価格’: [150, 100, 80, 300, 250], ‘在庫’: [50, 100, 80, 30, 40] }) print(“=== 保存するデータ ===”) print(df) # SQLiteデータベースに接続（ファイルがなければ自動作成） engine = create_engine(‘sqlite:///shop.db’) # テーブルに保存（テーブル名: products） df.to_sql(‘products’, engine, if_exists=’replace’, index=False) print(“\n✅ データをproductsテーブルに保存しました！”) # 保存したデータを確認 df_loaded = pd.read_sql(‘SELECT * FROM products’, engine) print(“\n=== 保存されたデータ ===”) print(df_loaded)

# ===== appendモードでデータ追加 ===== # 新しいデータ new_data = pd.DataFrame({ ‘商品ID’: [6, 7], ‘商品名’: [‘メロン’, ‘すいか’], ‘価格’: [500, 400], ‘在庫’: [20, 15] }) print(“=== 追加するデータ ===”) print(new_data) # 追加（appendモード） new_data.to_sql(‘products’, engine, if_exists=’append’, index=False) print(“\n✅ データを追加しました！”) # 確認 df_all = pd.read_sql(‘SELECT * FROM products’, engine) print(“\n=== 追加後のデータ（全件）===”) print(df_all)

# ===== データ型を明示的に指定 ===== from sqlalchemy.types import Integer, String df = pd.DataFrame({ ‘商品コード’: [‘A001’, ‘A002’, ‘A003’], ‘商品名’: [‘りんご’, ‘バナナ’, ‘みかん’], ‘価格’: [150, 100, 80], ‘説明’: [‘新鮮なりんご’, ‘甘いバナナ’, ‘ジューシーなみかん’] }) # データ型を定義 dtype_dict = { ‘商品コード’: String(10), # 文字列（最大10文字） ‘商品名’: String(100), # 文字列（最大100文字） ‘価格’: Integer, # 整数 ‘説明’: String(200) # 文字列（最大200文字） } # データ型を指定して保存 df.to_sql( ‘products_typed’, engine, if_exists=’replace’, index=False, dtype=dtype_dict ) print(“✅ データ型を指定して保存しました！”) # 確認 result = pd.read_sql(‘SELECT * FROM products_typed’, engine) print(result)

# ===== PostgreSQLへの接続 ===== from sqlalchemy import create_engine import os # 環境変数から接続情報を取得（推奨） DB_USER = os.getenv(‘DB_USER’, ‘postgres’) DB_PASS = os.getenv(‘DB_PASS’, ‘password’) DB_HOST = os.getenv(‘DB_HOST’, ‘localhost’) DB_PORT = os.getenv(‘DB_PORT’, ‘5432’) DB_NAME = os.getenv(‘DB_NAME’, ‘mydb’) # 接続文字列を作成 connection_string = f’postgresql://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}’ # エンジンを作成 engine = create_engine(connection_string) print(“✅ PostgreSQLに接続しました！”) # データを保存 df.to_sql(‘products’, engine, if_exists=’replace’, index=False) print(“✅ データを保存しました！”)

# ===== chunksizeを指定して大量データを保存 ===== import pandas as pd import numpy as np import time from sqlalchemy import create_engine engine = create_engine(‘sqlite:///large_data.db’) # 10万件のダミーデータを作成 print(“=== 10万件のデータを作成中… ===”) large_df = pd.DataFrame({ ‘商品ID’: range(1, 100001), ‘商品名’: [f’商品{i}’ for i in range(1, 100001)], ‘価格’: np.random.randint(100, 1000, 100000), ‘在庫’: np.random.randint(0, 100, 100000) }) print(f”データ件数: {len(large_df):,}件”) # 時間計測：chunksizeなし start = time.time() large_df.to_sql(‘products_no_chunk’, engine, if_exists=’replace’, index=False) time_no_chunk = time.time() – start print(f”\nchunksizeなし: {time_no_chunk:.2f}秒”) # 時間計測：chunksize=10000 start = time.time() large_df.to_sql(‘products_with_chunk’, engine, if_exists=’replace’, index=False, chunksize=10000) time_with_chunk = time.time() – start print(f”chunksize=10000: {time_with_chunk:.2f}秒”) # 件数確認 count = pd.read_sql(‘SELECT COUNT(*) as cnt FROM products_with_chunk’, engine) print(f”\n✅ 保存件数: {count[‘cnt’][0]:,}件”)

# ===== SQLiteでのUPSERT ===== import sqlite3 import pandas as pd conn = sqlite3.connect(‘upsert_demo.db’) cur = conn.cursor() # テーブル作成 cur.execute(”’ CREATE TABLE IF NOT EXISTS products ( product_id INTEGER PRIMARY KEY, product_name TEXT, price INTEGER, stock INTEGER ) ”’) # 初期データを挿入 initial_data = [ (1, ‘りんご’, 150, 50), (2, ‘バナナ’, 100, 100), (3, ‘みかん’, 80, 80) ] cur.executemany(”’ INSERT OR REPLACE INTO products VALUES (?, ?, ?, ?) ”’, initial_data) conn.commit() print(“=== 初期データ ===”) df = pd.read_sql(‘SELECT * FROM products’, conn) print(df) # UPSERTするデータ（ID=1は更新、ID=4は新規） upsert_data = [ (1, ‘りんご’, 160, 55), # 更新 (4, ‘ぶどう’, 300, 30) # 新規 ] # SQLiteのUPSERT: INSERT OR REPLACE cur.executemany(”’ INSERT OR REPLACE INTO products VALUES (?, ?, ?, ?) ”’, upsert_data) conn.commit() print(“\n=== UPSERT後 ===”) df = pd.read_sql(‘SELECT * FROM products ORDER BY product_id’, conn) print(df) conn.close()

# ===== PostgreSQLでのUPSERT ===== # PostgreSQLでは ON CONFLICT を使用 upsert_query = ”’ INSERT INTO products (product_id, product_name, price, stock) VALUES (%s, %s, %s, %s) ON CONFLICT (product_id) DO UPDATE SET product_name = EXCLUDED.product_name, price = EXCLUDED.price, stock = EXCLUDED.stock ”’ # EXCLUDEDは「挿入しようとした新しい値」を指す

# ===== トランザクションの基本 ===== import sqlite3 conn = sqlite3.connect(‘transaction_demo.db’) cur = conn.cursor() # テーブル作成 cur.execute(”’CREATE TABLE IF NOT EXISTS products ( id INTEGER PRIMARY KEY, name TEXT, stock INTEGER )”’) cur.execute(”’CREATE TABLE IF NOT EXISTS sales ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_id INTEGER, amount INTEGER )”’) cur.execute(‘DELETE FROM products’) cur.execute(‘DELETE FROM sales’) cur.execute(“INSERT INTO products VALUES (1, ‘りんご’, 100)”) conn.commit() print(“=== 初期状態 ===”) print(pd.read_sql(‘SELECT * FROM products’, conn)) try: # トランザクション開始 print(“\n=== トランザクション開始 ===”) # 処理1：在庫を減らす cur.execute(‘UPDATE products SET stock = stock – 10 WHERE id = 1’) print(“処理1: 在庫を10減らしました”) # 処理2：売上を記録 cur.execute(‘INSERT INTO sales (product_id, amount) VALUES (1, 10)’) print(“処理2: 売上を記録しました”) # 全て成功したらコミット conn.commit() print(“\n✅ トランザクション成功！（コミット）”) except Exception as e: # エラーが出たらロールバック conn.rollback() print(f”\n❌ エラー: {e}”) print(“トランザクションをロールバックしました”) print(“\n=== 最終状態 ===”) print(“【products】”) print(pd.read_sql(‘SELECT * FROM products’, conn)) print(“\n【sales】”) print(pd.read_sql(‘SELECT * FROM sales’, conn)) conn.close()