harness_zbt/backend/app/commands/reset_cache.py

from typing import Any, cast

from redis import Redis

from app.core.config import Settings


def clear_project_cache(settings: Settings | None = None) -> int:
    resolved_settings = settings or Settings()
    client = Redis.from_url(resolved_settings.redis_url, decode_responses=True)
    deleted = 0
    cursor = 0
    while True:
        cursor, keys = cast(
            tuple[int, list[str]],
            client.scan(
                cursor=cursor,
                match=f"{resolved_settings.redis_prefix}*",
                count=500,
            ),
        )
        if keys:
            deleted += cast(Any, client.delete(*keys))
        if cursor == 0:
            break
    client.close()
    return deleted


if __name__ == "__main__":
    count = clear_project_cache()
    print(f"已删除 {count} 个智保通第一阶段 Redis Key。")