本文目录导读：
我来分享一个Python工具类封装的完整案例,包含通用设计模式和最佳实践。
基础工具类设计
import os
import json
import yaml
import logging
import hashlib
from datetime import datetime
from typing import Any, Dict, List, Optional, Union
from pathlib import Path
class BaseTool:
"""工具基类，提供通用功能"""
def __init__(self, config: Optional[Dict] = None):
self.config = config or {}
self.logger = self._setup_logger()
def _setup_logger(self) -> logging.Logger:
"""配置日志器"""
logger = logging.getLogger(self.__class__.__name__)
logger.setLevel(logging.INFO)
if not logger.handlers:
handler = logging.StreamHandler()
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
handler.setFormatter(formatter)
logger.addHandler(handler)
return logger
def validate_config(self, required_keys: List[str]) -> bool:
"""验证配置完整性"""
missing_keys = [k for k in required_keys if k not in self.config]
if missing_keys:
self.logger.warning(f"Missing config keys: {missing_keys}")
return False
return True
class FileUtils(BaseTool):
"""文件操作工具类"""
def __init__(self, config: Optional[Dict] = None):
super().__init__(config)
self.supported_extensions = config.get('extensions', ['.txt', '.json', '.csv'])
def read_file(self, file_path: str, encoding: str = 'utf-8') -> str:
"""读取文件内容"""
try:
with open(file_path, 'r', encoding=encoding) as f:
return f.read()
except FileNotFoundError:
self.logger.error(f"File not found: {file_path}")
raise
except Exception as e:
self.logger.error(f"Error reading file: {e}")
raise
def write_file(self, file_path: str, content: str, mode: str = 'w') -> bool:
"""写入文件"""
try:
os.makedirs(os.path.dirname(file_path) or '.', exist_ok=True)
with open(file_path, mode, encoding='utf-8') as f:
f.write(content)
self.logger.info(f"File written successfully: {file_path}")
return True
except Exception as e:
self.logger.error(f"Error writing file: {e}")
return False
def json_reader(self, file_path: str) -> Dict:
"""读取JSON文件"""
content = self.read_file(file_path)
return json.loads(content)
def json_writer(self, file_path: str, data: Any, indent: int = 2) -> bool:
"""写入JSON文件"""
content = json.dumps(data, indent=indent, ensure_ascii=False)
return self.write_file(file_path, content)
def get_file_info(self, file_path: str) -> Dict:
"""获取文件信息"""
path = Path(file_path)
if not path.exists():
return {}
return {
'name': path.name,
'extension': path.suffix,
'size': path.stat().st_size,
'modified_time': datetime.fromtimestamp(path.stat().st_mtime).isoformat(),
'absolute_path': str(path.absolute())
}
class DataTransformer(BaseTool):
"""数据转换工具类"""
@staticmethod
def to_json(data: Any) -> str:
"""转换为JSON字符串"""
return json.dumps(data, ensure_ascii=False, default=str)
@staticmethod
def from_json(json_str: str) -> Any:
"""从JSON字符串转换"""
return json.loads(json_str)
@staticmethod
def to_yaml(data: Any) -> str:
"""转换为YAML字符串"""
return yaml.dump(data, default_flow_style=False, allow_unicode=True)
@staticmethod
def from_yaml(yaml_str: str) -> Any:
"""从YAML字符串转换"""
return yaml.safe_load(yaml_str)
def transform_list(self, data: List, operation: str, **kwargs) -> List:
"""列表转换操作"""
operations = {
'flatten': self._flatten_list,
'unique': lambda x: list(set(x)),
'sort': lambda x: sorted(x, **kwargs),
'filter': lambda x: [i for i in x if i not in kwargs.get('exclude', [])]
}
if operation in operations:
return operations[operation](data)
else:
self.logger.error(f"Unknown operation: {operation}")
return data
@staticmethod
def _flatten_list(nested_list: List) -> List:
"""展平嵌套列表"""
result = []
for item in nested_list:
if isinstance(item, list):
result.extend(DataTransformer._flatten_list(item))
else:
result.append(item)
return result
class SecurityUtils(BaseTool):
"""安全工具类"""
@staticmethod
def hash_password(password: str, salt: str = '') -> str:
"""密码哈希"""
combined = password + salt
return hashlib.sha256(combined.encode()).hexdigest()
@staticmethod
def md5_hash(text: str) -> str:
"""MD5哈希"""
return hashlib.md5(text.encode()).hexdigest()
@staticmethod
def generate_token(length: int = 32) -> str:
"""生成随机令牌"""
import secrets
return secrets.token_hex(length // 2)
def sanitize_filename(self, filename: str) -> str:
"""清理文件名，移除不安全字符"""
import re
# 移除非字母数字和.-_的字符
sanitized = re.sub(r'[^\w\.\-]', '_', filename)
# 限制长度
max_length = self.config.get('max_filename_length', 255)
return sanitized[:max_length]
class APIClient(BaseTool):
"""API客户端工具类"""
def __init__(self, config: Optional[Dict] = None):
super().__init__(config)
self.base_url = config.get('base_url', '')
self.timeout = config.get('timeout', 30)
self.retry_count = config.get('retry_count', 3)
def get(self, endpoint: str, params: Optional[Dict] = None) -> Dict:
"""GET请求"""
import requests
url = f"{self.base_url}/{endpoint.lstrip('/')}"
for attempt in range(self.retry_count):
try:
response = requests.get(url, params=params, timeout=self.timeout)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
self.logger.error(f"Request failed (attempt {attempt + 1}): {e}")
if attempt == self.retry_count - 1:
raise
def post(self, endpoint: str, data: Dict) -> Dict:
"""POST请求"""
import requests
url = f"{self.base_url}/{endpoint.lstrip('/')}"
try:
response = requests.post(url, json=data, timeout=self.timeout)
response.raise_for_status()
return response.json()
except requests.RequestException as e:
self.logger.error(f"POST request failed: {e}")
raise
高级工具类模式
class CacheUtils(BaseTool):
"""缓存工具类，支持多级缓存"""
def __init__(self, config: Optional[Dict] = None):
super().__init__(config)
self.cache = {}
self.cache_type = config.get('cache_type', 'memory')
self.max_size = config.get('max_cache_size', 100)
if self.cache_type == 'file':
self.cache_dir = Path(config.get('cache_dir', './cache'))
self.cache_dir.mkdir(exist_ok=True)
def get(self, key: str) -> Any:
"""获取缓存"""
if self.cache_type == 'memory':
return self.cache.get(key)
else:
return self._get_file_cache(key)
def set(self, key: str, value: Any, ttl: int = 3600) -> None:
"""设置缓存"""
if self.cache_type == 'memory':
self._set_memory_cache(key, value)
else:
self._set_file_cache(key, value)
self.logger.debug(f"Cache set for key: {key}")
def _set_memory_cache(self, key: str, value: Any) -> None:
"""内存缓存管理"""
if len(self.cache) >= self.max_size:
# LRU淘汰策略
self.cache.pop(next(iter(self.cache)))
self.cache[key] = value
def _get_file_cache(self, key: str) -> Optional[Any]:
"""文件缓存读取"""
cache_file = self.cache_dir / f"{hashlib.md5(key.encode()).hexdigest()}.json"
if cache_file.exists():
with open(cache_file, 'r') as f:
return json.load(f)
return None
def _set_file_cache(self, key: str, value: Any) -> None:
"""文件缓存写入"""
cache_file = self.cache_dir / f"{hashlib.md5(key.encode()).hexdigest()}.json"
with open(cache_file, 'w') as f:
json.dump(value, f)
def clear(self) -> None:
"""清空缓存"""
self.cache.clear()
self.logger.info("Cache cleared")
class ValidatorUtils(BaseTool):
"""数据验证工具类"""
@staticmethod
def validate_email(email: str) -> bool:
"""验证邮箱格式"""
import re
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
return bool(re.match(pattern, email))
@staticmethod
def validate_phone(phone: str) -> bool:
"""验证手机号格式"""
import re
pattern = r'^1[3-9]\d{9}$'
return bool(re.match(pattern, phone))
@staticmethod
def validate_url(url: str) -> bool:
"""验证URL格式"""
from urllib.parse import urlparse
try:
result = urlparse(url)
return all([result.scheme, result.netloc])
except:
return False
def validate_json_schema(self, data: Dict, schema: Dict) -> bool:
"""JSON Schema验证"""
try:
import jsonschema
jsonschema.validate(instance=data, schema=schema)
return True
except jsonschema.exceptions.ValidationError as e:
self.logger.error(f"Validation error: {e}")
return False
使用示例
def main():
"""使用示例"""
# 1. 文件工具使用
file_tool = FileUtils({
'extensions': ['.txt', '.json', '.csv', '.yaml']
})
# 写入文件
file_tool.write_file('config.json', '{"name": "test"}')
# 读取JSON
config = file_tool.json_reader('config.json')
print(f"Config: {config}")
# 2. 数据转换
transformer = DataTransformer()
nested_list = [1, [2, 3], [4, [5, 6]]]
flat_list = transformer.transform_list(nested_list, 'flatten')
print(f"Flattened: {flat_list}")
# 3. 安全工具
security = SecurityUtils()
password_hash = security.hash_password("my_password", "unique_salt")
token = security.generate_token()
print(f"Token: {token}")
# 4. 缓存使用
cache = CacheUtils({
'cache_type': 'memory',
'max_cache_size': 50
})
cache.set('user_42', {'name': 'Alice', 'age': 30})
user = cache.get('user_42')
print(f"Cached user: {user}")
# 5. 验证器使用
validator = ValidatorUtils()
email = "test@example.com"
print(f"Email valid: {validator.validate_email(email)}")
if __name__ == "__main__":
main()
最佳实践总结
设计原则
- 单一职责：每个工具类只负责一个功能领域
- 开闭原则：通过继承或配置扩展功能，不改动原有代码
- 依赖注入：通过配置参数控制行为
代码规范
- 使用类型注解提供更好的代码文档
- 添加适当的日志记录便于调试
- 提供异常处理机制
- 编写完整的文档字符串
扩展建议
- 添加单元测试覆盖核心功能
- 使用装饰器实现功能增强（如重试、缓存等）
- 考虑异步版本的工具类
- 实现插件机制支持自定义扩展
这样的工具类设计既保持灵活性，又提供了统一的调用接口,便于团队协作和代码维护。
