mcpField
R
redis

redis

github.com/thelobbiデータ・データベース2146 回インストール2 日前に更新
リポジトリ

redis について

Golden Armada AI Agent Fleet Platform向けのRedisスキルで、一般的なデータ処理と連携パターンを扱います。CLIとPythonの例を通じて、キャッシュ、セッション、メッセージング、レート制限、ロックを説明します。

主な機能

  1. 01Redis CLIの接続とコマンド
  2. 02同期・非同期Pythonの例
  3. 03キャッシュアサイドと無効化
  4. 04レート制限と分散ロック
  5. 05Pub/Subとセッション管理

活用シーン

  1. 01有効期限付きキャッシュの実装
  2. 02Redisハッシュによるセッション管理
  3. 03時間枠ごとのリクエスト制限
  4. 04分散ロックによる処理の調整

インストール

コマンドをコピーしてターミナルで実行します。

$npx skills add https://github.com/thelobbi/claude --skill redis

インストール後、Claude Code で/redisを入力して読み込まれたか確認します。

詳細

Redis スキル

Golden Armada AI Agent Fleet Platform 向けに、Redis の包括的な機能を提供します。

このスキルを使用するタイミング

次の作業を行うときに、このスキルを有効にします。

  • キャッシュの実装
  • セッション管理
  • Pub/Sub メッセージング
  • レート制限
  • 分散ロック

Redis CLI クイックリファレンス

接続

```bash

接続

redis-cli -h localhost -p 6379 redis-cli -h localhost -p 6379 -a password

接続をテスト

redis-cli ping ```

基本操作

```bash

文字列

SET key "value" SET key "value" EX 3600 # TTL を設定 GET key DEL key EXISTS key TTL key EXPIRE key 3600

ハッシュ

HSET user:1 name "John" age "30" HGET user:1 name HGETALL user:1 HDEL user:1 age

リスト

LPUSH queue "item1" RPUSH queue "item2" LPOP queue RPOP queue LRANGE queue 0 -1

セット

SADD tags "python" "redis" SMEMBERS tags SISMEMBER tags "python" SREM tags "python"

ソート済みセット

ZADD leaderboard 100 "player1" 200 "player2" ZRANGE leaderboard 0 -1 WITHSCORES ZRANK leaderboard "player1" ZINCRBY leaderboard 50 "player1" ```

Python Redis クライアント

```python import redis from redis import asyncio as aioredis

同期クライアント

r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

非同期クライアント

async_redis = aioredis.from_url("redis://localhost:6379", decode_responses=True)

基本操作

r.set('key', 'value', ex=3600) value = r.get('key') r.delete('key')

ハッシュ操作

r.hset('user:1', mapping={'name': 'John', 'age': '30'}) user = r.hgetall('user:1')

リスト操作

r.lpush('queue', 'item1', 'item2') items = r.lrange('queue', 0, -1) item = r.rpop('queue')

非同期操作

async def cache_get(key: str): async with aioredis.from_url("redis://localhost") as redis: return await redis.get(key) ```

キャッシュパターン

キャッシュアサイドパターン

```python async def get_agent(agent_id: str) -> dict: # まずキャッシュを確認 cache_key = f"agent:{agent_id}" cached = await redis.get(cache_key) if cached: return json.loads(cached)

# キャッシュミス時はデータベースから取得
agent = await db.get_agent(agent_id)
if agent:
    await redis.set(cache_key, json.dumps(agent), ex=3600)

return agent

async def update_agent(agent_id: str, data: dict) -> dict: # データベースを更新 agent = await db.update_agent(agent_id, data)

# キャッシュを無効化
cache_key = f"agent:{agent_id}"
await redis.delete(cache_key)

return agent

```

レート制限

```python async def rate_limit(key: str, limit: int, window: int) -> bool: """ スライディングウィンドウ方式のレートリミッターです。 リクエストが許可される場合は True を返します。 """ current = int(time.time()) window_key = f"rate:{key}:{current // window}"

async with redis.pipeline() as pipe:
    pipe.incr(window_key)
    pipe.expire(window_key, window * 2)
    results = await pipe.execute()

count = results[0]
return count <= limit

使用例

if not await rate_limit(f"user:{user_id}", limit=100, window=60): raise HTTPException(status_code=429, detail="レート制限を超過しました") ```

分散ロック

```python import uuid

class DistributedLock: def init(self, redis_client, key: str, timeout: int = 10): self.redis = redis_client self.key = f"lock:{key}" self.timeout = timeout self.token = str(uuid.uuid4())

async def __aenter__(self):
    while True:
        acquired = await self.redis.set(
            self.key, self.token, nx=True, ex=self.timeout
        )
        if acquired:
            return self
        await asyncio.sleep(0.1)

async def __aexit__(self, exc_type, exc_val, exc_tb):
    # ロックを所有している場合に限り解放
    script = """
    if redis.call("get", KEYS[1]) == ARGV[1] then
        return redis.call("del", KEYS[1])
    else
        return 0
    end
    """
    await self.redis.eval(script, 1, self.key, self.token)

使用方法

async with DistributedLock(redis, "resource:123"): await do_critical_work() ```

Pub/Sub

```python

Publisher

async def publish_event(channel: str, message: dict): await redis.publish(channel, json.dumps(message))

Subscriber

async def subscribe_events(): pubsub = redis.pubsub() await pubsub.subscribe("agent:events")

async for message in pubsub.listen():
    if message["type"] == "message":
        data = json.loads(message["data"])
        await handle_event(data)

```

セッション管理

```python from fastapi import Depends, HTTPException from fastapi.security import HTTPBearer import secrets

security = HTTPBearer()

async def create_session(user_id: str) -> str: session_id = secrets.token_urlsafe(32) session_key = f"session:{session_id}"

await redis.hset(session_key, mapping={
    "user_id": user_id,
    "created_at": datetime.utcnow().isoformat()
})
await redis.expire(session_key, 86400)  # 24 hours

return session_id

async def get_session(token: str = Depends(security)) -> dict: session_key = f"session:{token.credentials}" session = await redis.hgetall(session_key)

if not session:
    raise HTTPException(status_code=401, detail="Invalid session")

# Refresh TTL
await redis.expire(session_key, 86400)
return session

```

ベストプラクティス

  1. 本番環境では接続プーリングを使用します
  2. メモリの肥大化を防ぐため、すべてのキーに TTL を設定します
  3. バッチ操作にはパイプライン処理を使用します
  4. 接続の問題に対する適切なエラー処理を実装します
  5. INFO memory を使用して メモリ使用量を監視します
  6. アトミック操作には Lua スクリプトを使用します

コメント

Markdown 対応