SQLAlchemyによるデータベース接続

Pandas をデータベースに接続する理由

SQLAlchemy のインストール

# Install dependencies
# pip install sqlalchemy psycopg2-binary  # for PostgreSQL
# pip install sqlalchemy pymysql           # for MySQL
# sqlite3 is built into Python

import sqlalchemy as sa
import pandas as pd

print('SQLAlchemy version:', sa.__version__)

接続エンジンの作成

import sqlalchemy as sa
import os

# SQLite (file-based, no server needed)
sqlite_engine = sa.create_engine('sqlite:///mydata.db')

# PostgreSQL
# pg_url = 'postgresql://user:pass@localhost:5432/mydb'
# pg_engine = sa.create_engine(pg_url)

# From environment variable (safer)
# pg_engine = sa.create_engine(os.environ['DATABASE_URL'])

print(sqlite_engine)
print(type(sqlite_engine))

pd.read_sql_table() によるテーブルの読み込み

import pandas as pd
import sqlalchemy as sa

engine = sa.create_engine('sqlite:///sales.db')

# Read a full table
df = pd.read_sql_table('orders', con=engine)
print(df.shape)
print(df.dtypes)
print(df.head())

pd.read_sql_query() によるクエリの実行

import pandas as pd
import sqlalchemy as sa

engine = sa.create_engine('sqlite:///sales.db')

query = '''
    SELECT customer_id, SUM(amount) AS total_spent,
           COUNT(*) AS num_orders
    FROM orders
    WHERE status = 'completed'
    GROUP BY customer_id
    ORDER BY total_spent DESC
    LIMIT 100
'''

top_customers = pd.read_sql_query(query, con=engine)
print(top_customers.head())

安全なパラメーター化クエリ

import pandas as pd
import sqlalchemy as sa

engine = sa.create_engine('sqlite:///sales.db')

# Safe: parameterised query
params = {'status': 'completed', 'min_amount': 500.0}
query = sa.text(
    'SELECT * FROM orders WHERE status = :status AND amount > :min_amount'
)

with engine.connect() as conn:
    df = pd.read_sql_query(query, con=conn, params=params)
print(f'Loaded {len(df)} rows')

大規模なクエリ結果をチャンクで処理する

import pandas as pd
import sqlalchemy as sa

engine = sa.create_engine('postgresql://user:pass@host/db')

total = 0.0
count = 0

for chunk in pd.read_sql_query(
    'SELECT amount FROM orders',
    con=engine,
    chunksize=50000
):
    total += chunk['amount'].sum()
    count += len(chunk)

print(f'Mean amount: {total/count:.2f}')

接続のコンテキストマネージャー

import pandas as pd
import sqlalchemy as sa

engine = sa.create_engine('sqlite:///sales.db')

# Using context manager — connection always closed properly
with engine.connect() as conn:
    df = pd.read_sql_query(
        'SELECT * FROM products WHERE category = "Electronics"',
        con=conn
    )
    print(f'Products loaded: {len(df)}')
# Connection is automatically returned to the pool here

データベーススキーマの確認

import sqlalchemy as sa

engine = sa.create_engine('sqlite:///sales.db')
inspector = sa.inspect(engine)

# List all tables
tables = inspector.get_table_names()
print('Tables:', tables)

# Get columns for the 'orders' table
for col in inspector.get_columns('orders'):
    print(f'  {col["name"]}: {col["type"]}')

エンジンの終了とベストプラクティス

import sqlalchemy as sa

# Production-grade engine creation
engine = sa.create_engine(
    'postgresql://user:pass@host:5432/mydb',
    pool_size=5,          # max 5 persistent connections
    max_overflow=10,      # allow 10 temporary extra connections
    pool_pre_ping=True,   # verify connection before use
    connect_args={'connect_timeout': 10}
)

# ... run all your queries ...

# At the end of the application/script
engine.dispose()
print('Engine disposed')

read_sql と read_csv の速度比較

import pandas as pd
import sqlalchemy as sa
import time

engine = sa.create_engine('sqlite:///data.db')

# Database read with server-side filter
start = time.time()
df_sql = pd.read_sql_query(
    'SELECT * FROM transactions WHERE amount > 100 AND year = 2024',
    con=engine
)
print(f'SQL read: {time.time()-start:.3f}s, {len(df_sql):,} rows')

理解度チェック

レッスンのまとめ