91 lines
2.5 KiB
Python
91 lines
2.5 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
# @Time : 2022/3/28 15:28
|
||
# @Author : 李晓杰
|
||
"""
|
||
缓存文件处理
|
||
"""
|
||
|
||
import os
|
||
from typing import Any, Text, Union
|
||
from common.setting import ensure_path_sep
|
||
from utils.other_tools.exceptions import ValueNotFoundError
|
||
|
||
|
||
class Cache:
|
||
""" 设置、读取缓存 """
|
||
def __init__(self, filename: Union[Text, None]) -> None:
|
||
# 如果filename不为空，则操作指定文件内容
|
||
if filename:
|
||
self.path = ensure_path_sep("\\cache" + filename)
|
||
# 如果filename为None，则操作所有文件内容
|
||
else:
|
||
self.path = ensure_path_sep("\\cache")
|
||
|
||
def set_cache(self, key: Text, value: Any) -> None:
|
||
"""
|
||
设置缓存, 只支持设置单字典类型缓存数据, 缓存文件如以存在,则替换之前的缓存内容
|
||
:return:
|
||
"""
|
||
with open(self.path, 'w', encoding='utf-8') as file:
|
||
file.write(str({key: value}))
|
||
|
||
def set_caches(self, value: Any) -> None:
|
||
"""
|
||
设置多组缓存数据
|
||
:param value: 缓存内容
|
||
:return:
|
||
"""
|
||
with open(self.path, 'w', encoding='utf-8') as file:
|
||
file.write(str(value))
|
||
|
||
def get_cache(self) -> Any:
|
||
"""
|
||
获取缓存数据
|
||
:return:
|
||
"""
|
||
try:
|
||
with open(self.path, 'r', encoding='utf-8') as file:
|
||
return file.read()
|
||
except FileNotFoundError:
|
||
pass
|
||
|
||
def clean_cache(self) -> None:
|
||
"""删除所有缓存文件"""
|
||
|
||
if not os.path.exists(self.path):
|
||
raise FileNotFoundError(f"您要删除的缓存文件不存在 {self.path}")
|
||
os.remove(self.path)
|
||
|
||
@classmethod
|
||
def clean_all_cache(cls) -> None:
|
||
"""
|
||
清除所有缓存文件
|
||
:return:
|
||
"""
|
||
cache_path = ensure_path_sep("\\cache")
|
||
|
||
# 列出目录下所有文件，生成一个list
|
||
list_dir = os.listdir(cache_path)
|
||
for i in list_dir:
|
||
# 循环删除文件夹下得所有内容
|
||
os.remove(cache_path + i)
|
||
|
||
|
||
_cache_config = {}
|
||
|
||
|
||
class CacheHandler:
|
||
@staticmethod
|
||
def get_cache(cache_data):
|
||
try:
|
||
return _cache_config[cache_data]
|
||
except KeyError:
|
||
raise ValueNotFoundError(f"{cache_data}的缓存数据未找到，请检查是否将该数据存入缓存中")
|
||
|
||
@staticmethod
|
||
def update_cache(*, cache_name, value):
|
||
_cache_config[cache_name] = value
|
||
|
||
|