Python脚本如何批量更新缓存内容

wen python案例 2026-06-26 250

本文目录导读：

Python脚本如何批量更新缓存内容

使用Redis批量更新
使用文件缓存批量更新
使用内存字典缓存批量更新
混合缓存系统批量更新
配置文件批量更新

的Python方案：

使用Redis批量更新

import redis
import json
class RedisCacheUpdater:
    def __init__(self, host='localhost', port=6379, db=0):
        self.client = redis.Redis(host=host, port=port, db=db, decode_responses=True)
    def batch_update_by_keys(self, updates_dict):
        """
        批量更新指定key的值
        :param updates_dict: {key: new_value} 格式的字典
        """
        pipeline = self.client.pipeline()
        for key, value in updates_dict.items():
            pipeline.set(key, json.dumps(value) if isinstance(value, (dict, list)) else value)
        pipeline.execute()
        print(f"成功更新 {len(updates_dict)} 个缓存项")
    def batch_update_by_pattern(self, pattern, transform_func):
        """
        根据模式匹配批量更新
        :param pattern: key匹配模式，如 "user:*"
        :param transform_func: 转换函数，接收旧值返回新值
        """
        cursor = 0
        updated = 0
        while True:
            cursor, keys = self.client.scan(cursor, match=pattern, count=100)
            pipeline = self.client.pipeline()
            for key in keys:
                old_value = self.client.get(key)
                if old_value:
                    try:
                        old_value = json.loads(old_value)
                    except:
                        pass
                    new_value = transform_func(key, old_value)
                    pipeline.set(key, json.dumps(new_value) if isinstance(new_value, (dict, list)) else new_value)
                    updated += 1
            if pipeline:
                pipeline.execute()
            if cursor == 0:
                break
        print(f"根据模式 '{pattern}' 更新了 {updated} 个缓存项")
# 使用示例
updater = RedisCacheUpdater()
# 示例1：直接批量更新指定keys
updates = {
    "user:1": {"name": "张三", "status": "active"},
    "user:2": {"name": "李四", "status": "active"},
    "config:app": {"version": "2.0", "maintenance": False}
}
updater.batch_update_by_keys(updates)
# 示例2：根据模式批量更新
def update_user_status(key, old_value):
    if isinstance(old_value, dict):
        old_value["status"] = "active"
        old_value["updated_at"] = "2024-01-01"
    return old_value
updater.batch_update_by_pattern("user:*", update_user_status)

使用文件缓存批量更新

import os
import json
from pathlib import Path
import shutil
class FileCacheUpdater:
    def __init__(self, cache_dir):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(parents=True, exist_ok=True)
    def batch_update_from_dict(self, updates_dict):
        """
        从字典批量更新缓存文件
        :param updates_dict: {cache_name: content} 格式
        """
        for cache_name, content in updates_dict.items():
            cache_file = self.cache_dir / f"{cache_name}.json"
            with open(cache_file, 'w', encoding='utf-8') as f:
                json.dump(content, f, ensure_ascii=False, indent=2)
        print(f"成功更新 {len(updates_dict)} 个缓存文件")
    def batch_update_by_modification(self, source_dir, pattern="*.json"):
        """
        从源目录批量更新缓存
        :param source_dir: 源文件目录
        :param pattern: 文件匹配模式
        """
        source_path = Path(source_dir)
        for file_path in source_path.glob(pattern):
            # 读取新的缓存内容
            with open(file_path, 'r', encoding='utf-8') as f:
                content = json.load(f)
            # 写入缓存目录
            dest_file = self.cache_dir / file_path.name
            with open(dest_file, 'w', encoding='utf-8') as f:
                json.dump(content, f, ensure_ascii=False, indent=2)
        print(f"从 {source_dir} 更新了缓存文件")
    def batch_delete(self, patterns_or_keys):
        """
        批量删除缓存
        :param patterns_or_keys: 文件名列表或模式
        """
        for item in patterns_or_keys:
            for file_path in self.cache_dir.glob(item):
                file_path.unlink()
                print(f"删除缓存: {file_path.name}")
# 使用示例
updater = FileCacheUpdater("./cache")
# 示例1：直接更新多个缓存
updates = {
    "user_profile": {"name": "张三", "age": 30},
    "app_config": {"version": "2.0", "features": ["a", "b"]},
    "temp_data": [1, 2, 3, 4]
}
updater.batch_update_from_dict(updates)
# 示例2：从源目录批量更新
updater.batch_update_by_modification("./new_cache_data")
# 示例3：批量删除过期缓存
updater.batch_delete(["temp_*", "session_*"])

使用内存字典缓存批量更新

import time
import threading
from datetime import datetime
class InMemoryCacheUpdater:
    def __init__(self):
        self._cache = {}
        self._expiry = {}
        self._lock = threading.Lock()
    def batch_update(self, items, ttl=None):
        """
        批量更新缓存
        :param items: {key: value} 格式
        :param ttl: 过期时间（秒）
        """
        with self._lock:
            expire_time = time.time() + ttl if ttl else None
            for key, value in items.items():
                self._cache[key] = value
                if expire_time:
                    self._expiry[key] = expire_time
        print(f"批量更新 {len(items)} 个缓存项")
    def batch_update_with_condition(self, condition_func, transform_func):
        """
        根据条件批量更新
        :param condition_func: 条件函数，接收(key, value)返回布尔值
        :param transform_func: 转换函数，接收(key, value)返回新value
        """
        with self._lock:
            updated = 0
            keys_to_update = []
            for key, value in self._cache.items():
                if condition_func(key, value):
                    keys_to_update.append(key)
            for key in keys_to_update:
                old_value = self._cache[key]
                new_value = transform_func(key, old_value)
                self._cache[key] = new_value
                updated += 1
        print(f"根据条件更新了 {updated} 个缓存项")
    def batch_delete_expired(self):
        """批量删除过期缓存"""
        with self._lock:
            now = time.time()
            expired_keys = [
                key for key, expire_time in self._expiry.items()
                if expire_time <= now
            ]
            for key in expired_keys:
                del self._cache[key]
                del self._expiry[key]
        if expired_keys:
            print(f"清理了 {len(expired_keys)} 个过期缓存项")
        return expired_keys
    def get_all(self):
        """获取所有缓存"""
        with self._lock:
            return dict(self._cache)
# 使用示例
cache = InMemoryCacheUpdater()
# 示例1：批量更新
cache.batch_update({
    "user:1": {"name": "张三", "role": "admin"},
    "user:2": {"name": "李四", "role": "user"},
    "config": {"theme": "dark", "language": "zh"}
}, ttl=3600)
# 示例2：根据条件批量更新
def is_admin(key, value):
    return isinstance(value, dict) and value.get("role") == "admin"
def update_admin_role(key, value):
    value["permissions"] = ["read", "write", "delete"]
    return value
cache.batch_update_with_condition(is_admin, update_admin_role)

混合缓存系统批量更新

class HybridCacheUpdater:
    """支持多层级缓存批量更新"""
    def __init__(self):
        self.cache_layers = []
    def add_cache_layer(self, cache_obj, name):
        """添加缓存层"""
        self.cache_layers.append({"obj": cache_obj, "name": name})
    def batch_update_all_layers(self, updates_dict):
        """
        在所有缓存层批量更新
        """
        for layer in self.cache_layers:
            print(f"更新缓存层: {layer['name']}")
            layer['obj'].batch_update(updates_dict)
    def batch_update_selected_layers(self, updates_dict, layer_names):
        """
        在指定缓存层批量更新
        """
        for layer in self.cache_layers:
            if layer['name'] in layer_names:
                print(f"更新缓存层: {layer['name']}")
                layer['obj'].batch_update(updates_dict)
# 使用示例
# 创建各种缓存实例
redis_cache = RedisCacheUpdater()
file_cache = FileCacheUpdater("./cache")
memory_cache = InMemoryCacheUpdater()
# 组合成混合缓存
hybrid = HybridCacheUpdater()
hybrid.add_cache_layer(redis_cache, "redis")
hybrid.add_cache_layer(file_cache, "file")
hybrid.add_cache_layer(memory_cache, "memory")
# 批量更新所有缓存层
updates = {
    "common:config": {"version": "3.0", "updated": True}
}
hybrid.batch_update_all_layers(updates)

配置文件批量更新

import configparser
import yaml
class ConfigCacheUpdater:
    """批量更新配置文件缓存"""
    @staticmethod
    def update_ini_config(config_file, updates):
        """
        更新INI配置文件
        :param config_file: 配置文件路径
        :param updates: {section: {option: value}} 格式
        """
        config = configparser.ConfigParser()
        config.read(config_file, encoding='utf-8')
        for section, options in updates.items():
            if not config.has_section(section):
                config.add_section(section)
            for option, value in options.items():
                config.set(section, option, str(value))
        with open(config_file, 'w', encoding='utf-8') as f:
            config.write(f)
        print(f"INI配置文件 {config_file} 已更新 {len(updates)} 个节")
    @staticmethod
    def update_yaml_config(config_file, updates):
        """
        更新YAML配置文件
        :param updates: 嵌套字典格式的更新内容
        """
        def deep_update(original, updates):
            for key, value in updates.items():
                if isinstance(value, dict) and key in original:
                    original[key] = deep_update(original.get(key, {}), value)
                else:
                    original[key] = value
            return original
        with open(config_file, 'r', encoding='utf-8') as f:
            config = yaml.safe_load(f)
        config = deep_update(config, updates)
        with open(config_file, 'w', encoding='utf-8') as f:
            yaml.dump(config, f, default_flow_style=False, allow_unicode=True)
        print(f"YAML配置文件 {config_file} 已更新")
# 使用示例
updater = ConfigCacheUpdater()
# 更新INI配置
updater.update_ini_config("config.ini", {
    "database": {"host": "localhost", "port": "3306"},
    "cache": {"ttl": "3600", "size": "100MB"}
})
# 更新YAML配置
updater.update_yaml_config("config.yaml", {
    "server": {
        "host": "0.0.0.0",
        "port": 8080
    },
    "database": {
        "pool_size": 10
    }
})

这些方案覆盖了常见的缓存更新场景，可以根据实际需求选择合适的方案,需要我详细解释某个方案或提供更多定制化的批量更新逻辑吗？

本文地址： https://www.bmjup.com/post/9565.html