PythonでORMを使ったデータベースクラスタリングとシャーディングの実践

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# クラスタ内の各DBの接続情報
DB_URLS = [
    'postgresql://user1:password1@host1/db1',
    'postgresql://user2:password2@host2/db2',
    # ...
]

engines = [create_engine(url) for url in DB_URLS]

# ランダムにDBを選ぶシンプルなロジック
import random
selected_engine = random.choice(engines)

Session = sessionmaker(bind=selected_engine)
session = Session()
# 以下、DB操作

# 同上のライブラリのインポートは省略

def choose_shard(key):
    # keyに基づいてシャードを選ぶロジック
    return DB_URLS[key % len(DB_URLS)]

key = 42  # 任意のキー
selected_url = choose_shard(key)

selected_engine = create_engine(selected_url)
Session = sessionmaker(bind=selected_engine)
session = Session()
# 以下、DB操作

# キャッシュ用ライブラリのインポート
import redis

cache = redis.Redis(host='localhost', port=6379, db=0)

def choose_db_with_cache(key):
    cached_db = cache.get(key)
    if cached_db:
        return cached_db.decode('utf-8')
    else:
        selected_db = random.choice(DB_URLS)
        cache.set(key, selected_db)
        return selected_db

# 使用例
db_url = choose_db_with_cache('some_key')
engine = create_engine(db_url)
Session = sessionmaker(bind=engine)
session = Session()
# 以下、DB操作

# 同上のライブラリのインポートは省略

def migrate_data(old_key, new_key):
    old_db = choose_shard(old_key)
    new_db = choose_shard(new_key)
    
    if old_db != new_db:
        old_engine = create_engine(old_db)
        new_engine = create_engine(new_db)
        # データの移行処理（省略）